Make it possible to run Load test using Deployments or ReplicaSets

test/e2e/framework/util.go

@@ -2622,9 +2622,29 @@ func RemoveTaintOffNode(c clientset.Interface, nodeName string, taint v1.Taint)
 	}
 }
 
-func ScaleRC(clientset clientset.Interface, internalClientset internalclientset.Interface, ns, name string, size uint, wait bool) error {
-	By(fmt.Sprintf("Scaling replication controller %s in namespace %s to %d", name, ns, size))
-	scaler, err := kubectl.ScalerFor(api.Kind("ReplicationController"), internalClientset)
+func getScalerForKind(internalClientset internalclientset.Interface, kind schema.GroupKind) (kubectl.Scaler, error) {
+	switch kind {
+	case api.Kind("ReplicationController"):
+		return kubectl.ScalerFor(api.Kind("ReplicationController"), internalClientset)
+	case extensionsinternal.Kind("ReplicaSet"):
+		return kubectl.ScalerFor(extensionsinternal.Kind("ReplicaSet"), internalClientset)
+	case extensionsinternal.Kind("Deployment"):
+		return kubectl.ScalerFor(extensionsinternal.Kind("Deployment"), internalClientset)
+	default:
+		return nil, fmt.Errorf("Unsupported kind for getting Scaler: %v", kind)
+	}
+}
+
+func ScaleResource(
+	clientset clientset.Interface,
+	internalClientset internalclientset.Interface,
+	ns, name string,
+	size uint,
+	wait bool,
+	kind schema.GroupKind,
+) error {
+	By(fmt.Sprintf("Scaling %v %s in namespace %s to %d", kind, name, ns, size))
+	scaler, err := getScalerForKind(internalClientset, kind)
 	if err != nil {
 		return err
 	}
@@ -2636,51 +2656,32 @@ func ScaleRC(clientset clientset.Interface, internalClientset internalclientset.
 	if !wait {
 		return nil
 	}
-	return WaitForRCPodsRunning(clientset, ns, name)
+	return WaitForControlledPodsRunning(clientset, ns, name, kind)
 }
 
 // Wait up to 10 minutes for pods to become Running.
-func WaitForRCPodsRunning(c clientset.Interface, ns, rcName string) error {
-	rc, err := c.Core().ReplicationControllers(ns).Get(rcName)
+func WaitForControlledPodsRunning(c clientset.Interface, ns, name string, kind schema.GroupKind) error {
+	rtObject, err := getRuntimeObjectForKind(c, kind, ns, name)
+	if err != nil {
+		return err
+	}
+	selector, err := getSelectorFromRuntimeObject(rtObject)
 	if err != nil {
 		return err
 	}
-	selector := labels.SelectorFromSet(labels.Set(rc.Spec.Selector))
 	err = testutils.WaitForPodsWithLabelRunning(c, ns, selector)
 	if err != nil {
-		return fmt.Errorf("Error while waiting for replication controller %s pods to be running: %v", rcName, err)
+		return fmt.Errorf("Error while waiting for replication controller %s pods to be running: %v", name, err)
 	}
 	return nil
 }
 
+func ScaleRC(clientset clientset.Interface, internalClientset internalclientset.Interface, ns, name string, size uint, wait bool) error {
+	return ScaleResource(clientset, internalClientset, ns, name, size, wait, api.Kind("ReplicationController"))
+}
+
 func ScaleDeployment(clientset clientset.Interface, internalClientset internalclientset.Interface, ns, name string, size uint, wait bool) error {
-	By(fmt.Sprintf("Scaling Deployment %s in namespace %s to %d", name, ns, size))
-	scaler, err := kubectl.ScalerFor(extensionsinternal.Kind("Deployment"), internalClientset)
-	if err != nil {
-		return err
-	}
-	waitForScale := kubectl.NewRetryParams(5*time.Second, 1*time.Minute)
-	waitForReplicas := kubectl.NewRetryParams(5*time.Second, 5*time.Minute)
-	if err = scaler.Scale(ns, name, size, nil, waitForScale, waitForReplicas); err != nil {
-		return fmt.Errorf("error while scaling Deployment %s to %d replicas: %v", name, size, err)
-	}
-	if !wait {
-		return nil
-	}
-	return WaitForDeploymentPodsRunning(clientset, ns, name)
-}
-
-func WaitForDeploymentPodsRunning(c clientset.Interface, ns, name string) error {
-	deployment, err := c.Extensions().Deployments(ns).Get(name)
-	if err != nil {
-		return err
-	}
-	selector := labels.SelectorFromSet(labels.Set(deployment.Spec.Selector.MatchLabels))
-	err = testutils.WaitForPodsWithLabelRunning(c, ns, selector)
-	if err != nil {
-		return fmt.Errorf("Error while waiting for Deployment %s pods to be running: %v", name, err)
-	}
-	return nil
+	return ScaleResource(clientset, internalClientset, ns, name, size, wait, extensionsinternal.Kind("Deployment"))
 }
 
 // Returns true if all the specified pods are scheduled, else returns false.
@@ -2760,26 +2761,102 @@ func WaitForPodsWithLabelRunningReady(c clientset.Interface, ns string, label la
 	return pods, err
 }
 
-// DeleteRCAndPods a Replication Controller and all pods it spawned
-func DeleteRCAndPods(clientset clientset.Interface, internalClientset internalclientset.Interface, ns, name string) error {
-	By(fmt.Sprintf("deleting replication controller %s in namespace %s", name, ns))
-	rc, err := clientset.Core().ReplicationControllers(ns).Get(name)
+func getRuntimeObjectForKind(c clientset.Interface, kind schema.GroupKind, ns, name string) (runtime.Object, error) {
+	switch kind {
+	case api.Kind("ReplicationController"):
+		return c.Core().ReplicationControllers(ns).Get(name)
+	case extensionsinternal.Kind("ReplicaSet"):
+		return c.Extensions().ReplicaSets(ns).Get(name)
+	case extensionsinternal.Kind("Deployment"):
+		return c.Extensions().Deployments(ns).Get(name)
+	default:
+		return nil, fmt.Errorf("Unsupported kind when getting runtime object: %v", kind)
+	}
+}
+
+func deleteResource(c clientset.Interface, kind schema.GroupKind, ns, name string, deleteOption *v1.DeleteOptions) error {
+	switch kind {
+	case api.Kind("ReplicationController"):
+		return c.Core().ReplicationControllers(ns).Delete(name, deleteOption)
+	case extensionsinternal.Kind("ReplicaSet"):
+		return c.Extensions().ReplicaSets(ns).Delete(name, deleteOption)
+	case extensionsinternal.Kind("Deployment"):
+		return c.Extensions().Deployments(ns).Delete(name, deleteOption)
+	default:
+		return fmt.Errorf("Unsupported kind when deleting: %v", kind)
+	}
+}
+
+func getSelectorFromRuntimeObject(obj runtime.Object) (labels.Selector, error) {
+	switch typed := obj.(type) {
+	case *v1.ReplicationController:
+		return labels.SelectorFromSet(typed.Spec.Selector), nil
+	case *extensions.ReplicaSet:
+		return metav1.LabelSelectorAsSelector(typed.Spec.Selector)
+	case *extensions.Deployment:
+		return metav1.LabelSelectorAsSelector(typed.Spec.Selector)
+	default:
+		return nil, fmt.Errorf("Unsupported kind when getting selector: %v", obj)
+	}
+}
+
+func getReplicasFromRuntimeObject(obj runtime.Object) (int32, error) {
+	switch typed := obj.(type) {
+	case *v1.ReplicationController:
+		if typed.Spec.Replicas != nil {
+			return *typed.Spec.Replicas, nil
+		}
+		return 0, nil
+	case *extensions.ReplicaSet:
+		if typed.Spec.Replicas != nil {
+			return *typed.Spec.Replicas, nil
+		}
+		return 0, nil
+	case *extensions.Deployment:
+		if typed.Spec.Replicas != nil {
+			return *typed.Spec.Replicas, nil
+		}
+		return 0, nil
+	default:
+		return -1, fmt.Errorf("Unsupported kind when getting number of replicas: %v", obj)
+	}
+}
+
+func getReaperForKind(internalClientset internalclientset.Interface, kind schema.GroupKind) (kubectl.Reaper, error) {
+	switch kind {
+	case api.Kind("ReplicationController"):
+		return kubectl.ReaperFor(api.Kind("ReplicationController"), internalClientset)
+	case extensionsinternal.Kind("ReplicaSet"):
+		return kubectl.ReaperFor(extensionsinternal.Kind("ReplicaSet"), internalClientset)
+	case extensionsinternal.Kind("Deployment"):
+		return kubectl.ReaperFor(extensionsinternal.Kind("Deployment"), internalClientset)
+	default:
+		return nil, fmt.Errorf("Unsupported kind: %v", kind)
+	}
+}
+
+// DeleteResourceAndPods deletes a given resource and all pods it spawned
+func DeleteResourceAndPods(clientset clientset.Interface, internalClientset internalclientset.Interface, kind schema.GroupKind, ns, name string) error {
+	By(fmt.Sprintf("deleting %v %s in namespace %s", kind, name, ns))
+
+	rtObject, err := getRuntimeObjectForKind(clientset, kind, ns, name)
 	if err != nil {
 		if apierrs.IsNotFound(err) {
-			Logf("RC %s was already deleted: %v", name, err)
+			Logf("%v %s not found: %v", kind, name, err)
 			return nil
 		}
 		return err
 	}
-	reaper, err := kubectl.ReaperForReplicationController(internalClientset.Core(), 10*time.Minute)
+	selector, err := getSelectorFromRuntimeObject(rtObject)
 	if err != nil {
-		if apierrs.IsNotFound(err) {
-			Logf("RC %s was already deleted: %v", name, err)
-			return nil
-		}
 		return err
 	}
-	ps, err := podStoreForRC(clientset, rc)
+	reaper, err := getReaperForKind(internalClientset, kind)
+	if err != nil {
+		return err
+	}
+
+	ps, err := podStoreForSelector(clientset, ns, selector)
 	if err != nil {
 		return err
 	}
@@ -2787,20 +2864,20 @@ func DeleteRCAndPods(clientset clientset.Interface, internalClientset internalcl
 	startTime := time.Now()
 	err = reaper.Stop(ns, name, 0, nil)
 	if apierrs.IsNotFound(err) {
-		Logf("RC %s was already deleted: %v", name, err)
+		Logf("%v %s was already deleted: %v", kind, name, err)
 		return nil
 	}
 	if err != nil {
-		return fmt.Errorf("error while stopping RC: %s: %v", name, err)
+		return fmt.Errorf("error while stopping %v: %s: %v", kind, name, err)
 	}
-	deleteRCTime := time.Now().Sub(startTime)
-	Logf("Deleting RC %s took: %v", name, deleteRCTime)
+	deleteTime := time.Now().Sub(startTime)
+	Logf("Deleting %v %s took: %v", kind, name, deleteTime)
 	err = waitForPodsInactive(ps, 10*time.Millisecond, 10*time.Minute)
 	if err != nil {
 		return fmt.Errorf("error while waiting for pods to become inactive %s: %v", name, err)
 	}
-	terminatePodTime := time.Now().Sub(startTime) - deleteRCTime
-	Logf("Terminating RC %s pods took: %v", name, terminatePodTime)
+	terminatePodTime := time.Now().Sub(startTime) - deleteTime
+	Logf("Terminating %v %s pods took: %v", kind, name, terminatePodTime)
 	// this is to relieve namespace controller's pressure when deleting the
 	// namespace after a test.
 	err = waitForPodsGone(ps, 10*time.Second, 10*time.Minute)
@@ -2810,57 +2887,75 @@ func DeleteRCAndPods(clientset clientset.Interface, internalClientset internalcl
 	return nil
 }
 
-// DeleteRCAndWaitForGC deletes only the Replication Controller and waits for GC to delete the pods.
-func DeleteRCAndWaitForGC(c clientset.Interface, ns, name string) error {
-	By(fmt.Sprintf("deleting replication controller %s in namespace %s, will wait for the garbage collector to delete the pods", name, ns))
-	rc, err := c.Core().ReplicationControllers(ns).Get(name)
+func DeleteRCAndPods(clientset clientset.Interface, internalClientset internalclientset.Interface, ns, name string) error {
+	return DeleteResourceAndPods(clientset, internalClientset, api.Kind("ReplicationController"), ns, name)
+}
+
+// DeleteResourceAndWaitForGC deletes only given resource and waits for GC to delete the pods.
+func DeleteResourceAndWaitForGC(c clientset.Interface, kind schema.GroupKind, ns, name string) error {
+	By(fmt.Sprintf("deleting %v %s in namespace %s, will wait for the garbage collector to delete the pods", kind, name, ns))
+
+	rtObject, err := getRuntimeObjectForKind(c, kind, ns, name)
 	if err != nil {
 		if apierrs.IsNotFound(err) {
-			Logf("RC %s was already deleted: %v", name, err)
+			Logf("%v %s not found: %v", kind, name, err)
 			return nil
 		}
 		return err
 	}
-	ps, err := podStoreForRC(c, rc)
+	selector, err := getSelectorFromRuntimeObject(rtObject)
 	if err != nil {
 		return err
 	}
+	replicas, err := getReplicasFromRuntimeObject(rtObject)
+	if err != nil {
+		return err
+	}
+
+	ps, err := podStoreForSelector(c, ns, selector)
+	if err != nil {
+		return err
+	}
+
 	defer ps.Stop()
 	startTime := time.Now()
 	falseVar := false
 	deleteOption := &v1.DeleteOptions{OrphanDependents: &falseVar}
-	err = c.Core().ReplicationControllers(ns).Delete(name, deleteOption)
+	err = deleteResource(c, kind, ns, name, deleteOption)
 	if err != nil && apierrs.IsNotFound(err) {
-		Logf("RC %s was already deleted: %v", name, err)
+		Logf("%v %s was already deleted: %v", kind, name, err)
 		return nil
 	}
 	if err != nil {
 		return err
 	}
-	deleteRCTime := time.Now().Sub(startTime)
-	Logf("Deleting RC %s took: %v", name, deleteRCTime)
+	deleteTime := time.Now().Sub(startTime)
+	Logf("Deleting %v %s took: %v", kind, name, deleteTime)
+
 	var interval, timeout time.Duration
 	switch {
-	case *(rc.Spec.Replicas) < 100:
+	case replicas < 100:
 		interval = 100 * time.Millisecond
-	case *(rc.Spec.Replicas) < 1000:
+	case replicas < 1000:
 		interval = 1 * time.Second
 	default:
 		interval = 10 * time.Second
 	}
-	if *(rc.Spec.Replicas) < 5000 {
+	if replicas < 5000 {
 		timeout = 10 * time.Minute
 	} else {
-		timeout = time.Duration(*(rc.Spec.Replicas)/gcThroughput) * time.Second
+		timeout = time.Duration(replicas/gcThroughput) * time.Second
 		// gcThroughput is pretty strict now, add a bit more to it
 		timeout = timeout + 3*time.Minute
 	}
+
 	err = waitForPodsInactive(ps, interval, timeout)
 	if err != nil {
 		return fmt.Errorf("error while waiting for pods to become inactive %s: %v", name, err)
 	}
-	terminatePodTime := time.Now().Sub(startTime) - deleteRCTime
-	Logf("Terminating RC %s pods took: %v", name, terminatePodTime)
+	terminatePodTime := time.Now().Sub(startTime) - deleteTime
+	Logf("Terminating %v %s pods took: %v", kind, name, terminatePodTime)
+
 	err = waitForPodsGone(ps, interval, 10*time.Minute)
 	if err != nil {
 		return fmt.Errorf("error while waiting for pods gone %s: %v", name, err)
@@ -2868,11 +2963,15 @@ func DeleteRCAndWaitForGC(c clientset.Interface, ns, name string) error {
 	return nil
 }
 
-// podStoreForRC creates a PodStore that monitors pods belong to the rc. It
-// waits until the reflector does a List() before returning.
-func podStoreForRC(c clientset.Interface, rc *v1.ReplicationController) (*testutils.PodStore, error) {
-	labels := labels.SelectorFromSet(rc.Spec.Selector)
-	ps := testutils.NewPodStore(c, rc.Namespace, labels, fields.Everything())
+// DeleteRCAndWaitForGC deletes only the Replication Controller and waits for GC to delete the pods.
+func DeleteRCAndWaitForGC(c clientset.Interface, ns, name string) error {
+	return DeleteResourceAndWaitForGC(c, api.Kind("ReplicationController"), ns, name)
+}
+
+// podStoreForSelector creates a PodStore that monitors pods from given namespace matching given selector.
+// It waits until the reflector does a List() before returning.
+func podStoreForSelector(c clientset.Interface, ns string, selector labels.Selector) (*testutils.PodStore, error) {
+	ps := testutils.NewPodStore(c, ns, selector, fields.Everything())
 	err := wait.Poll(1*time.Second, 2*time.Minute, func() (bool, error) {
 		if len(ps.Reflector.LastSyncResourceVersion()) != 0 {
 			return true, nil
@@ -4312,7 +4411,7 @@ func ScaleRCByLabels(clientset clientset.Interface, internalClientset internalcl
 			return err
 		}
 		if replicas == 0 {
-			ps, err := podStoreForRC(clientset, rc)
+			ps, err := podStoreForSelector(clientset, rc.Namespace, labels.SelectorFromSet(rc.Spec.Selector))
 			if err != nil {
 				return err
 			}