Merge pull request #31000 from mksalawa/fix_pods_from_all_namespaces

Automatic merge from submit-queue

Fix getting pods from all namespaces

**What this PR does / why we need it**:
Use Heapster handler for pods from all namespaces (added in the new version).

Depends on #30993

pkg/kubectl/cmd/top_pod_test.go

@@ -30,9 +30,58 @@ import (
 	"net/url"
 )
 
+func TestTopPodAllNamespacesMetrics(t *testing.T) {
+	initTestErrorHandler(t)
+	metrics := testPodMetricsData()
+	firstTestNamespace := "testnamespace"
+	secondTestNamespace := "secondtestns"
+	thirdTestNamespace := "thirdtestns"
+	metrics.Items[0].Namespace = firstTestNamespace
+	metrics.Items[1].Namespace = secondTestNamespace
+	metrics.Items[2].Namespace = thirdTestNamespace
+
+	expectedPath := fmt.Sprintf("%s/%s/pods", baseMetricsAddress, metricsApiVersion)
+
+	f, tf, _, ns := NewAPIFactory()
+	tf.Printer = &testPrinter{}
+	tf.Client = &fake.RESTClient{
+		NegotiatedSerializer: ns,
+		Client: fake.CreateHTTPClient(func(req *http.Request) (*http.Response, error) {
+			switch p, m := req.URL.Path, req.Method; {
+			case p == expectedPath && m == "GET":
+				body, err := marshallBody(metrics)
+				if err != nil {
+					t.Errorf("unexpected error: %v", err)
+				}
+				return &http.Response{StatusCode: 200, Header: defaultHeader(), Body: body}, nil
+			default:
+				t.Fatalf("unexpected request: %#v\nGot URL: %#v\nExpected path: %#v", req, req.URL, expectedPath)
+				return nil, nil
+			}
+		}),
+	}
+	tf.Namespace = firstTestNamespace
+	tf.ClientConfig = &restclient.Config{ContentConfig: restclient.ContentConfig{GroupVersion: &unversioned.GroupVersion{Version: "v1"}}}
+	buf := bytes.NewBuffer([]byte{})
+
+	cmd := NewCmdTopPod(f, buf)
+	cmd.Flags().Set("all-namespaces", "true")
+	cmd.Run(cmd, []string{})
+
+	// Check the presence of pod names and namespaces in the output.
+	result := buf.String()
+	for _, m := range metrics.Items {
+		if !strings.Contains(result, m.Name) {
+			t.Errorf("missing metrics for %s: \n%s", m.Name, result)
+		}
+		if !strings.Contains(result, m.Namespace) {
+			t.Errorf("missing metrics for %s/%s: \n%s", m.Namespace, m.Name, result)
+		}
+	}
+}
+
 func TestTopPodAllInNamespaceMetrics(t *testing.T) {
 	initTestErrorHandler(t)
-	// TODO(magorzata): refactor to pods/ path after updating heapster version
 	metrics := testPodMetricsData()
 	testNamespace := "testnamespace"
 	nonTestNamespace := "anothernamespace"
@@ -47,7 +96,7 @@ func TestTopPodAllInNamespaceMetrics(t *testing.T) {
 		ListMeta: metrics.ListMeta,
 		Items:    metrics.Items[2:],
 	}
-	for _, m := range expectedMetrics.Items {
+	for _, m := range nonExpectedMetrics.Items {
 		m.Namespace = nonTestNamespace
 	}
 	expectedPath := fmt.Sprintf("%s/%s/namespaces/%s/pods", baseMetricsAddress, metricsApiVersion, testNamespace)

pkg/kubectl/metricsutil/metrics_client.go

@@ -38,7 +38,7 @@ const (
 var (
 	prefix       = "/apis"
 	groupVersion = fmt.Sprintf("%s/%s", metricsGv.Group, metricsGv.Version)
-	MetricsRoot  = fmt.Sprintf("%s/%s", prefix, groupVersion)
+	metricsRoot  = fmt.Sprintf("%s/%s", prefix, groupVersion)
 
 	// TODO: get this from metrics api once it's finished
 	metricsGv = unversioned.GroupVersion{Group: "metrics", Version: "v1alpha1"}
@@ -67,6 +67,9 @@ func DefaultHeapsterMetricsClient(client *client.Client) *HeapsterMetricsClient
 }
 
 func podMetricsUrl(namespace string, name string) (string, error) {
+	if namespace == api.NamespaceAll {
+		return fmt.Sprintf("%s/pods", metricsRoot), nil
+	}
 	errs := validation.ValidateNamespaceName(namespace, false)
 	if len(errs) > 0 {
 		message := fmt.Sprintf("invalid namespace: %s - %v", namespace, errs)
@@ -79,7 +82,7 @@ func podMetricsUrl(namespace string, name string) (string, error) {
 			return "", errors.New(message)
 		}
 	}
-	return fmt.Sprintf("%s/namespaces/%s/pods/%s", MetricsRoot, namespace, name), nil
+	return fmt.Sprintf("%s/namespaces/%s/pods/%s", metricsRoot, namespace, name), nil
 }
 
 func nodeMetricsUrl(name string) (string, error) {
@@ -90,7 +93,7 @@ func nodeMetricsUrl(name string) (string, error) {
 			return "", errors.New(message)
 		}
 	}
-	return fmt.Sprintf("%s/nodes/%s", MetricsRoot, name), nil
+	return fmt.Sprintf("%s/nodes/%s", metricsRoot, name), nil
 }
 
 func (cli *HeapsterMetricsClient) GetNodeMetrics(nodeName string, selector string) ([]metrics_api.NodeMetrics, error) {
@@ -123,47 +126,34 @@ func (cli *HeapsterMetricsClient) GetNodeMetrics(nodeName string, selector strin
 }
 
 func (cli *HeapsterMetricsClient) GetPodMetrics(namespace string, podName string, allNamespaces bool, selector string) ([]metrics_api.PodMetrics, error) {
-	// TODO: extend Master Metrics API with getting pods from all namespaces
-	// instead of aggregating the results here
-	namespaces := make([]string, 0)
 	if allNamespaces {
-		list, err := cli.Client.Namespaces().List(api.ListOptions{})
-		if err != nil {
-			return []metrics_api.PodMetrics{}, err
-		}
-		for _, ns := range list.Items {
-			namespaces = append(namespaces, ns.Name)
-		}
-	} else {
-		namespaces = append(namespaces, namespace)
+		namespace = api.NamespaceAll
+	}
+	path, err := podMetricsUrl(namespace, podName)
+	if err != nil {
+		return []metrics_api.PodMetrics{}, err
 	}
-
 	params := map[string]string{"labelSelector": selector}
 	allMetrics := make([]metrics_api.PodMetrics, 0)
-	for _, ns := range namespaces {
-		path, err := podMetricsUrl(ns, podName)
+
+	resultRaw, err := GetHeapsterMetrics(cli, path, params)
+	if err != nil {
+		return []metrics_api.PodMetrics{}, err
+	}
+	if len(podName) == 0 {
+		metrics := metrics_api.PodMetricsList{}
+		err = json.Unmarshal(resultRaw, &metrics)
 		if err != nil {
-			return []metrics_api.PodMetrics{}, err
+			return []metrics_api.PodMetrics{}, fmt.Errorf("failed to unmarshall heapster response: %v", err)
 		}
-		resultRaw, err := GetHeapsterMetrics(cli, path, params)
+		allMetrics = append(allMetrics, metrics.Items...)
+	} else {
+		var singleMetric metrics_api.PodMetrics
+		err = json.Unmarshal(resultRaw, &singleMetric)
 		if err != nil {
-			return []metrics_api.PodMetrics{}, err
-		}
-		if len(podName) == 0 {
-			metrics := metrics_api.PodMetricsList{}
-			err = json.Unmarshal(resultRaw, &metrics)
-			if err != nil {
-				return []metrics_api.PodMetrics{}, fmt.Errorf("failed to unmarshall heapster response: %v", err)
-			}
-			allMetrics = append(allMetrics, metrics.Items...)
-		} else {
-			var singleMetric metrics_api.PodMetrics
-			err = json.Unmarshal(resultRaw, &singleMetric)
-			if err != nil {
-				return []metrics_api.PodMetrics{}, fmt.Errorf("failed to unmarshall heapster response: %v", err)
-			}
-			allMetrics = append(allMetrics, singleMetric)
+			return []metrics_api.PodMetrics{}, fmt.Errorf("failed to unmarshall heapster response: %v", err)
 		}
+		allMetrics = append(allMetrics, singleMetric)
 	}
 	return allMetrics, nil
 }