Move k8s utils to pkg/util

test/integration/util/kubernetes.go → pkg/util/kubernetes.go

@@ -18,7 +18,6 @@ package util
 
 import (
 	"fmt"
-	"testing"
 	"time"
 
 	"github.com/pkg/errors"
@@ -30,11 +29,13 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/kubernetes/cmd/kubeadm/app/constants"
 
 	apierrs "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/pkg/api/v1"
 
+	"github.com/golang/glog"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/clientcmd"
 )
@@ -109,31 +110,36 @@ func StartPods(c kubernetes.Interface, namespace string, pod v1.Pod, waitForRunn
 // Wait up to 10 minutes for all matching pods to become Running and at least one
 // matching pod exists.
 func WaitForPodsWithLabelRunning(c kubernetes.Interface, ns string, label labels.Selector) error {
-	running := false
-	PodStore := NewPodStore(c, ns, label, fields.Everything())
-	defer PodStore.Stop()
-waitLoop:
-	for start := time.Now(); time.Since(start) < 10*time.Minute; time.Sleep(250 * time.Millisecond) {
-		pods := PodStore.List()
-		if len(pods) == 0 {
-			continue waitLoop
+	lastKnownPodNumber := -1
+	return wait.PollImmediate(constants.APICallRetryInterval, time.Minute*10, func() (bool, error) {
+		listOpts := metav1.ListOptions{LabelSelector: label.String()}
+		pods, err := c.CoreV1().Pods(ns).List(listOpts)
+		if err != nil {
+			glog.Infof("error getting Pods with label selector %q [%v]\n", label.String(), err)
+			return false, nil
+		}
+
+		if lastKnownPodNumber != len(pods.Items) {
+			glog.Infof("Found %d Pods for label selector %s\n", len(pods.Items), label.String())
+			lastKnownPodNumber = len(pods.Items)
+		}
+
+		if len(pods.Items) == 0 {
+			return false, nil
 		}
-		for _, p := range pods {
-			if p.Status.Phase != v1.PodRunning {
-				continue waitLoop
+
+		for _, pod := range pods.Items {
+			if pod.Status.Phase != v1.PodRunning {
+				return false, nil
 			}
 		}
-		running = true
-		break
-	}
-	if !running {
-		return fmt.Errorf("Timeout while waiting for pods with labels %q to be running", label.String())
-	}
-	return nil
+
+		return true, nil
+	})
 }
 
 // WaitForRCToStabilize waits till the RC has a matching generation/replica count between spec and status.
-func WaitForRCToStabilize(t *testing.T, c kubernetes.Interface, ns, name string, timeout time.Duration) error {
+func WaitForRCToStabilize(c kubernetes.Interface, ns, name string, timeout time.Duration) error {
 	options := metav1.ListOptions{FieldSelector: fields.Set{
 		"metadata.name":      name,
 		"metadata.namespace": ns,
@@ -154,7 +160,7 @@ func WaitForRCToStabilize(t *testing.T, c kubernetes.Interface, ns, name string,
 				*(rc.Spec.Replicas) == rc.Status.Replicas {
 				return true, nil
 			}
-			t.Logf("Waiting for rc %s to stabilize, generation %v observed generation %v spec.replicas %d status.replicas %d",
+			glog.Infof("Waiting for rc %s to stabilize, generation %v observed generation %v spec.replicas %d status.replicas %d",
 				name, rc.Generation, rc.Status.ObservedGeneration, *(rc.Spec.Replicas), rc.Status.Replicas)
 		}
 		return false, nil
@@ -163,21 +169,21 @@ func WaitForRCToStabilize(t *testing.T, c kubernetes.Interface, ns, name string,
 }
 
 // WaitForService waits until the service appears (exist == true), or disappears (exist == false)
-func WaitForService(t *testing.T, c kubernetes.Interface, namespace, name string, exist bool, interval, timeout time.Duration) error {
+func WaitForService(c kubernetes.Interface, namespace, name string, exist bool, interval, timeout time.Duration) error {
 	err := wait.PollImmediate(interval, timeout, func() (bool, error) {
 		_, err := c.Core().Services(namespace).Get(name, metav1.GetOptions{})
 		switch {
 		case err == nil:
-			t.Logf("Service %s in namespace %s found.", name, namespace)
+			glog.Infof("Service %s in namespace %s found.", name, namespace)
 			return exist, nil
 		case apierrs.IsNotFound(err):
-			t.Logf("Service %s in namespace %s disappeared.", name, namespace)
+			glog.Infof("Service %s in namespace %s disappeared.", name, namespace)
 			return !exist, nil
 		case !IsRetryableAPIError(err):
-			t.Logf("Non-retryable failure while getting service.")
+			glog.Infof("Non-retryable failure while getting service.")
 			return false, err
 		default:
-			t.Logf("Get service %s in namespace %s failed: %v", name, namespace, err)
+			glog.Infof("Get service %s in namespace %s failed: %v", name, namespace, err)
 			return false, nil
 		}
 	})
@@ -189,9 +195,9 @@ func WaitForService(t *testing.T, c kubernetes.Interface, namespace, name string
 }
 
 //WaitForServiceEndpointsNum waits until the amount of endpoints that implement service to expectNum.
-func WaitForServiceEndpointsNum(t *testing.T, c kubernetes.Interface, namespace, serviceName string, expectNum int, interval, timeout time.Duration) error {
+func WaitForServiceEndpointsNum(c kubernetes.Interface, namespace, serviceName string, expectNum int, interval, timeout time.Duration) error {
 	return wait.Poll(interval, timeout, func() (bool, error) {
-		t.Logf("Waiting for amount of service:%s endpoints to be %d", serviceName, expectNum)
+		glog.Infof("Waiting for amount of service:%s endpoints to be %d", serviceName, expectNum)
 		list, err := c.Core().Endpoints(namespace).List(metav1.ListOptions{})
 		if err != nil {
 			return false, err

test/integration/addons_test.go

@@ -27,17 +27,18 @@ import (
 	"time"
 
 	"k8s.io/apimachinery/pkg/labels"
+	pkgutil "k8s.io/minikube/pkg/util"
 	"k8s.io/minikube/test/integration/util"
 )
 
 func testAddons(t *testing.T) {
 	t.Parallel()
-	client, err := util.GetClient()
+	client, err := pkgutil.GetClient()
 	if err != nil {
 		t.Fatalf("Could not get kubernetes client: %s", err)
 	}
 	selector := labels.SelectorFromSet(labels.Set(map[string]string{"component": "kube-addon-manager"}))
-	if err := util.WaitForPodsWithLabelRunning(client, "kube-system", selector); err != nil {
+	if err := pkgutil.WaitForPodsWithLabelRunning(client, "kube-system", selector); err != nil {
 		t.Errorf("Error waiting for addon manager to be up")
 	}
 }

test/integration/mount_test.go

@@ -28,6 +28,7 @@ import (
 	"time"
 
 	"k8s.io/apimachinery/pkg/labels"
+	pkgutil "k8s.io/minikube/pkg/util"
 	"k8s.io/minikube/test/integration/util"
 )
 
@@ -76,12 +77,12 @@ func testMounting(t *testing.T) {
 		t.Fatal("mountTest failed with error:", err)
 	}
 
-	client, err := util.GetClient()
+	client, err := pkgutil.GetClient()
 	if err != nil {
 		t.Fatalf("getting kubernetes client: %s", err)
 	}
 	selector := labels.SelectorFromSet(labels.Set(map[string]string{"integration-test": "busybox-mount"}))
-	if err := util.WaitForPodsWithLabelRunning(client, "default", selector); err != nil {
+	if err := pkgutil.WaitForPodsWithLabelRunning(client, "default", selector); err != nil {
 		t.Fatalf("Error waiting for busybox mount pod to be up: %s", err)
 	}
 

test/integration/util/util.go

@@ -217,46 +217,46 @@ func (k *KubectlRunner) DeleteNamespace(namespace string) error {
 }
 
 func WaitForBusyboxRunning(t *testing.T, namespace string) error {
-	client, err := GetClient()
+	client, err := commonutil.GetClient()
 	if err != nil {
 		return errors.Wrap(err, "getting kubernetes client")
 	}
 	selector := labels.SelectorFromSet(labels.Set(map[string]string{"integration-test": "busybox"}))
-	return WaitForPodsWithLabelRunning(client, namespace, selector)
+	return commonutil.WaitForPodsWithLabelRunning(client, namespace, selector)
 }
 
 func WaitForDNSRunning(t *testing.T) error {
-	client, err := GetClient()
+	client, err := commonutil.GetClient()
 	if err != nil {
 		return errors.Wrap(err, "getting kubernetes client")
 	}
 
 	selector := labels.SelectorFromSet(labels.Set(map[string]string{"k8s-app": "kube-dns"}))
-	if err := WaitForPodsWithLabelRunning(client, "kube-system", selector); err != nil {
+	if err := commonutil.WaitForPodsWithLabelRunning(client, "kube-system", selector); err != nil {
 		return errors.Wrap(err, "waiting for kube-dns pods")
 	}
 
-	if err := WaitForService(t, client, "kube-system", "kube-dns", true, time.Millisecond*500, time.Minute*10); err != nil {
+	if err := commonutil.WaitForService(client, "kube-system", "kube-dns", true, time.Millisecond*500, time.Minute*10); err != nil {
 		t.Errorf("Error waiting for kube-dns service to be up")
 	}
 
 	return nil
 }
 
 func WaitForDashboardRunning(t *testing.T) error {
-	client, err := GetClient()
+	client, err := commonutil.GetClient()
 	if err != nil {
 		return errors.Wrap(err, "getting kubernetes client")
 	}
-	if err := WaitForRCToStabilize(t, client, "kube-system", "kubernetes-dashboard", time.Minute*10); err != nil {
+	if err := commonutil.WaitForRCToStabilize(client, "kube-system", "kubernetes-dashboard", time.Minute*10); err != nil {
 		return errors.Wrap(err, "waiting for dashboard RC to stabilize")
 	}
 
-	if err := WaitForService(t, client, "kube-system", "kubernetes-dashboard", true, time.Millisecond*500, time.Minute*10); err != nil {
+	if err := commonutil.WaitForService(client, "kube-system", "kubernetes-dashboard", true, time.Millisecond*500, time.Minute*10); err != nil {
 		return errors.Wrap(err, "waiting for dashboard service to be up")
 	}
 
-	if err := WaitForServiceEndpointsNum(t, client, "kube-system", "kubernetes-dashboard", 1, time.Second*3, time.Minute*10); err != nil {
+	if err := commonutil.WaitForServiceEndpointsNum(client, "kube-system", "kubernetes-dashboard", 1, time.Second*3, time.Minute*10); err != nil {
 		return errors.Wrap(err, "waiting for one dashboard endpoint to be up")
 	}