Merge pull request #60243 from MaciekPytel/hpa_api_ext_imp

Automatic merge from submit-queue (batch tested with PRs 60433, 59982, 59128, 60243, 60440). If you want to cherry-pick this change to another branch, please follow the instructions <a href="https://github.com/kubernetes/community/blob/master/contributors/devel/cherry-picks.md">here</a>.

Implement external metric in HPA

This implement the changes to HPA introduced in https://github.com/kubernetes/kubernetes/pull/60096

cmd/kube-controller-manager/app/BUILD

@@ -129,6 +129,7 @@ go_library(
         "//vendor/k8s.io/client-go/util/cert:go_default_library",
         "//vendor/k8s.io/metrics/pkg/client/clientset_generated/clientset/typed/metrics/v1beta1:go_default_library",
         "//vendor/k8s.io/metrics/pkg/client/custom_metrics:go_default_library",
+        "//vendor/k8s.io/metrics/pkg/client/external_metrics:go_default_library",
     ],
 )
 

cmd/kube-controller-manager/app/autoscaling.go

@@ -31,6 +31,7 @@ import (
 	"k8s.io/kubernetes/pkg/controller/podautoscaler/metrics"
 	resourceclient "k8s.io/metrics/pkg/client/clientset_generated/clientset/typed/metrics/v1beta1"
 	"k8s.io/metrics/pkg/client/custom_metrics"
+	"k8s.io/metrics/pkg/client/external_metrics"
 )
 
 func startHPAController(ctx ControllerContext) (bool, error) {
@@ -51,6 +52,7 @@ func startHPAControllerWithRESTClient(ctx ControllerContext) (bool, error) {
 	metricsClient := metrics.NewRESTMetricsClient(
 		resourceclient.NewForConfigOrDie(clientConfig),
 		custom_metrics.NewForConfigOrDie(clientConfig),
+		external_metrics.NewForConfigOrDie(clientConfig),
 	)
 	return startHPAControllerWithMetricsClient(ctx, metricsClient)
 }

pkg/controller/podautoscaler/BUILD

@@ -82,10 +82,12 @@ go_test(
         "//vendor/k8s.io/client-go/testing:go_default_library",
         "//vendor/k8s.io/heapster/metrics/api/v1/types:go_default_library",
         "//vendor/k8s.io/metrics/pkg/apis/custom_metrics/v1beta1:go_default_library",
+        "//vendor/k8s.io/metrics/pkg/apis/external_metrics/v1beta1:go_default_library",
         "//vendor/k8s.io/metrics/pkg/apis/metrics/v1alpha1:go_default_library",
         "//vendor/k8s.io/metrics/pkg/apis/metrics/v1beta1:go_default_library",
         "//vendor/k8s.io/metrics/pkg/client/clientset_generated/clientset/fake:go_default_library",
         "//vendor/k8s.io/metrics/pkg/client/custom_metrics/fake:go_default_library",
+        "//vendor/k8s.io/metrics/pkg/client/external_metrics/fake:go_default_library",
     ],
 )
 

pkg/controller/podautoscaler/horizontal.go

@@ -299,6 +299,45 @@ func (a *HorizontalController) computeReplicasForMetrics(hpa *autoscalingv2.Hori
 					},
 				}
 			}
+		case autoscalingv2.ExternalMetricSourceType:
+			if metricSpec.External.TargetAverageValue != nil {
+				replicaCountProposal, utilizationProposal, timestampProposal, err = a.replicaCalc.GetExternalPerPodMetricReplicas(currentReplicas, metricSpec.External.TargetAverageValue.MilliValue(), metricSpec.External.MetricName, hpa.Namespace, metricSpec.External.MetricSelector)
+				if err != nil {
+					a.eventRecorder.Event(hpa, v1.EventTypeWarning, "FailedGetExternalMetric", err.Error())
+					setCondition(hpa, autoscalingv2.ScalingActive, v1.ConditionFalse, "FailedGetExternalMetric", "the HPA was unable to compute the replica count: %v", err)
+					return 0, "", nil, time.Time{}, fmt.Errorf("failed to get %s external metric: %v", metricSpec.External.MetricName, err)
+				}
+				metricNameProposal = fmt.Sprintf("external metric %s(%+v)", metricSpec.External.MetricName, metricSpec.External.MetricSelector)
+				statuses[i] = autoscalingv2.MetricStatus{
+					Type: autoscalingv2.ExternalMetricSourceType,
+					External: &autoscalingv2.ExternalMetricStatus{
+						MetricSelector:      metricSpec.External.MetricSelector,
+						MetricName:          metricSpec.External.MetricName,
+						CurrentAverageValue: resource.NewMilliQuantity(utilizationProposal, resource.DecimalSI),
+					},
+				}
+			} else if metricSpec.External.TargetValue != nil {
+				replicaCountProposal, utilizationProposal, timestampProposal, err = a.replicaCalc.GetExternalMetricReplicas(currentReplicas, metricSpec.External.TargetValue.MilliValue(), metricSpec.External.MetricName, hpa.Namespace, metricSpec.External.MetricSelector)
+				if err != nil {
+					a.eventRecorder.Event(hpa, v1.EventTypeWarning, "FailedGetExternalMetric", err.Error())
+					setCondition(hpa, autoscalingv2.ScalingActive, v1.ConditionFalse, "FailedGetExternalMetric", "the HPA was unable to compute the replica count: %v", err)
+					return 0, "", nil, time.Time{}, fmt.Errorf("failed to get external metric %s: %v", metricSpec.External.MetricName, err)
+				}
+				metricNameProposal = fmt.Sprintf("external metric %s(%+v)", metricSpec.External.MetricName, metricSpec.External.MetricSelector)
+				statuses[i] = autoscalingv2.MetricStatus{
+					Type: autoscalingv2.ExternalMetricSourceType,
+					External: &autoscalingv2.ExternalMetricStatus{
+						MetricSelector: metricSpec.External.MetricSelector,
+						MetricName:     metricSpec.External.MetricName,
+						CurrentValue:   *resource.NewMilliQuantity(utilizationProposal, resource.DecimalSI),
+					},
+				}
+			} else {
+				errMsg := "invalid external metric source: neither a value target nor an average value target was set"
+				a.eventRecorder.Event(hpa, v1.EventTypeWarning, "FailedGetExternalMetric", errMsg)
+				setCondition(hpa, autoscalingv2.ScalingActive, v1.ConditionFalse, "FailedGetExternalMetric", "the HPA was unable to compute the replica count: %v", err)
+				return 0, "", nil, time.Time{}, fmt.Errorf(errMsg)
+			}
 		default:
 			errMsg := fmt.Sprintf("unknown metric source type %q", string(metricSpec.Type))
 			a.eventRecorder.Event(hpa, v1.EventTypeWarning, "InvalidMetricSourceType", errMsg)

pkg/controller/podautoscaler/horizontal_test.go

@@ -42,9 +42,11 @@ import (
 	"k8s.io/kubernetes/pkg/controller"
 	"k8s.io/kubernetes/pkg/controller/podautoscaler/metrics"
 	cmapi "k8s.io/metrics/pkg/apis/custom_metrics/v1beta1"
+	emapi "k8s.io/metrics/pkg/apis/external_metrics/v1beta1"
 	metricsapi "k8s.io/metrics/pkg/apis/metrics/v1beta1"
 	metricsfake "k8s.io/metrics/pkg/client/clientset_generated/clientset/fake"
 	cmfake "k8s.io/metrics/pkg/client/custom_metrics/fake"
+	emfake "k8s.io/metrics/pkg/client/external_metrics/fake"
 
 	"github.com/stretchr/testify/assert"
 
@@ -145,7 +147,7 @@ func init() {
 	scaleUpLimitFactor = 8
 }
 
-func (tc *testCase) prepareTestClient(t *testing.T) (*fake.Clientset, *metricsfake.Clientset, *cmfake.FakeCustomMetricsClient, *scalefake.FakeScaleClient) {
+func (tc *testCase) prepareTestClient(t *testing.T) (*fake.Clientset, *metricsfake.Clientset, *cmfake.FakeCustomMetricsClient, *emfake.FakeExternalMetricsClient, *scalefake.FakeScaleClient) {
 	namespace := "test-namespace"
 	hpaName := "test-hpa"
 	podNamePrefix := "test-pod"
@@ -523,7 +525,34 @@ func (tc *testCase) prepareTestClient(t *testing.T) (*fake.Clientset, *metricsfa
 		return true, metrics, nil
 	})
 
-	return fakeClient, fakeMetricsClient, fakeCMClient, fakeScaleClient
+	fakeEMClient := &emfake.FakeExternalMetricsClient{}
+
+	fakeEMClient.AddReactor("list", "*", func(action core.Action) (handled bool, ret runtime.Object, err error) {
+		tc.Lock()
+		defer tc.Unlock()
+
+		listAction, wasList := action.(core.ListAction)
+		if !wasList {
+			return true, nil, fmt.Errorf("expected a list action, got %v instead", action)
+		}
+
+		metrics := &emapi.ExternalMetricValueList{}
+
+		assert.Equal(t, "qps", listAction.GetResource().Resource, "the metric name requested should have been qps, as specified in the metric spec")
+
+		for _, level := range tc.reportedLevels {
+			metric := emapi.ExternalMetricValue{
+				Timestamp:  metav1.Time{Time: time.Now()},
+				MetricName: "qps",
+				Value:      *resource.NewMilliQuantity(int64(level), resource.DecimalSI),
+			}
+			metrics.Items = append(metrics.Items, metric)
+		}
+
+		return true, metrics, nil
+	})
+
+	return fakeClient, fakeMetricsClient, fakeCMClient, fakeEMClient, fakeScaleClient
 }
 
 func (tc *testCase) verifyResults(t *testing.T) {
@@ -538,7 +567,7 @@ func (tc *testCase) verifyResults(t *testing.T) {
 }
 
 func (tc *testCase) setupController(t *testing.T) (*HorizontalController, informers.SharedInformerFactory) {
-	testClient, testMetricsClient, testCMClient, testScaleClient := tc.prepareTestClient(t)
+	testClient, testMetricsClient, testCMClient, testEMClient, testScaleClient := tc.prepareTestClient(t)
 	if tc.testClient != nil {
 		testClient = tc.testClient
 	}
@@ -554,6 +583,7 @@ func (tc *testCase) setupController(t *testing.T) (*HorizontalController, inform
 	metricsClient := metrics.NewRESTMetricsClient(
 		testMetricsClient.MetricsV1beta1(),
 		testCMClient,
+		testEMClient,
 	)
 
 	eventClient := &fake.Clientset{}
@@ -822,6 +852,48 @@ func TestScaleUpCMObject(t *testing.T) {
 	tc.runTest(t)
 }
 
+func TestScaleUpCMExternal(t *testing.T) {
+	tc := testCase{
+		minReplicas:     2,
+		maxReplicas:     6,
+		initialReplicas: 3,
+		desiredReplicas: 4,
+		metricsTarget: []autoscalingv2.MetricSpec{
+			{
+				Type: autoscalingv2.ExternalMetricSourceType,
+				External: &autoscalingv2.ExternalMetricSource{
+					MetricSelector: &metav1.LabelSelector{},
+					MetricName:     "qps",
+					TargetValue:    resource.NewMilliQuantity(6666, resource.DecimalSI),
+				},
+			},
+		},
+		reportedLevels: []uint64{8600},
+	}
+	tc.runTest(t)
+}
+
+func TestScaleUpPerPodCMExternal(t *testing.T) {
+	tc := testCase{
+		minReplicas:     2,
+		maxReplicas:     6,
+		initialReplicas: 3,
+		desiredReplicas: 4,
+		metricsTarget: []autoscalingv2.MetricSpec{
+			{
+				Type: autoscalingv2.ExternalMetricSourceType,
+				External: &autoscalingv2.ExternalMetricSource{
+					MetricSelector:     &metav1.LabelSelector{},
+					MetricName:         "qps",
+					TargetAverageValue: resource.NewMilliQuantity(2222, resource.DecimalSI),
+				},
+			},
+		},
+		reportedLevels: []uint64{8600},
+	}
+	tc.runTest(t)
+}
+
 func TestScaleDown(t *testing.T) {
 	tc := testCase{
 		minReplicas:         2,
@@ -886,6 +958,48 @@ func TestScaleDownCMObject(t *testing.T) {
 	tc.runTest(t)
 }
 
+func TestScaleDownCMExternal(t *testing.T) {
+	tc := testCase{
+		minReplicas:     2,
+		maxReplicas:     6,
+		initialReplicas: 5,
+		desiredReplicas: 3,
+		metricsTarget: []autoscalingv2.MetricSpec{
+			{
+				Type: autoscalingv2.ExternalMetricSourceType,
+				External: &autoscalingv2.ExternalMetricSource{
+					MetricSelector: &metav1.LabelSelector{},
+					MetricName:     "qps",
+					TargetValue:    resource.NewMilliQuantity(14400, resource.DecimalSI),
+				},
+			},
+		},
+		reportedLevels: []uint64{8600},
+	}
+	tc.runTest(t)
+}
+
+func TestScaleDownPerPodCMExternal(t *testing.T) {
+	tc := testCase{
+		minReplicas:     2,
+		maxReplicas:     6,
+		initialReplicas: 5,
+		desiredReplicas: 3,
+		metricsTarget: []autoscalingv2.MetricSpec{
+			{
+				Type: autoscalingv2.ExternalMetricSourceType,
+				External: &autoscalingv2.ExternalMetricSource{
+					MetricSelector:     &metav1.LabelSelector{},
+					MetricName:         "qps",
+					TargetAverageValue: resource.NewMilliQuantity(3000, resource.DecimalSI),
+				},
+			},
+		},
+		reportedLevels: []uint64{8600},
+	}
+	tc.runTest(t)
+}
+
 func TestScaleDownIgnoresUnreadyPods(t *testing.T) {
 	tc := testCase{
 		minReplicas:          2,
@@ -979,6 +1093,58 @@ func TestToleranceCMObject(t *testing.T) {
 	tc.runTest(t)
 }
 
+func TestToleranceCMExternal(t *testing.T) {
+	tc := testCase{
+		minReplicas:     2,
+		maxReplicas:     6,
+		initialReplicas: 4,
+		desiredReplicas: 4,
+		metricsTarget: []autoscalingv2.MetricSpec{
+			{
+				Type: autoscalingv2.ExternalMetricSourceType,
+				External: &autoscalingv2.ExternalMetricSource{
+					MetricSelector: &metav1.LabelSelector{},
+					MetricName:     "qps",
+					TargetValue:    resource.NewMilliQuantity(8666, resource.DecimalSI),
+				},
+			},
+		},
+		reportedLevels: []uint64{8600},
+		expectedConditions: statusOkWithOverrides(autoscalingv2.HorizontalPodAutoscalerCondition{
+			Type:   autoscalingv2.AbleToScale,
+			Status: v1.ConditionTrue,
+			Reason: "ReadyForNewScale",
+		}),
+	}
+	tc.runTest(t)
+}
+
+func TestTolerancePerPodCMExternal(t *testing.T) {
+	tc := testCase{
+		minReplicas:     2,
+		maxReplicas:     6,
+		initialReplicas: 4,
+		desiredReplicas: 4,
+		metricsTarget: []autoscalingv2.MetricSpec{
+			{
+				Type: autoscalingv2.ExternalMetricSourceType,
+				External: &autoscalingv2.ExternalMetricSource{
+					MetricSelector:     &metav1.LabelSelector{},
+					MetricName:         "qps",
+					TargetAverageValue: resource.NewMilliQuantity(2200, resource.DecimalSI),
+				},
+			},
+		},
+		reportedLevels: []uint64{8600},
+		expectedConditions: statusOkWithOverrides(autoscalingv2.HorizontalPodAutoscalerCondition{
+			Type:   autoscalingv2.AbleToScale,
+			Status: v1.ConditionTrue,
+			Reason: "ReadyForNewScale",
+		}),
+	}
+	tc.runTest(t)
+}
+
 func TestMinReplicas(t *testing.T) {
 	tc := testCase{
 		minReplicas:         2,
@@ -1268,7 +1434,7 @@ func TestConditionInvalidSelectorMissing(t *testing.T) {
 		},
 	}
 
-	_, _, _, testScaleClient := tc.prepareTestClient(t)
+	_, _, _, _, testScaleClient := tc.prepareTestClient(t)
 	tc.testScaleClient = testScaleClient
 
 	testScaleClient.PrependReactor("get", "replicationcontrollers", func(action core.Action) (handled bool, ret runtime.Object, err error) {
@@ -1313,7 +1479,7 @@ func TestConditionInvalidSelectorUnparsable(t *testing.T) {
 		},
 	}
 
-	_, _, _, testScaleClient := tc.prepareTestClient(t)
+	_, _, _, _, testScaleClient := tc.prepareTestClient(t)
 	tc.testScaleClient = testScaleClient
 
 	testScaleClient.PrependReactor("get", "replicationcontrollers", func(action core.Action) (handled bool, ret runtime.Object, err error) {
@@ -1374,7 +1540,7 @@ func TestConditionFailedGetMetrics(t *testing.T) {
 			reportedCPURequests: []resource.Quantity{resource.MustParse("0.1"), resource.MustParse("0.1"), resource.MustParse("0.1")},
 			useMetricsAPI:       true,
 		}
-		_, testMetricsClient, testCMClient, _ := tc.prepareTestClient(t)
+		_, testMetricsClient, testCMClient, _, _ := tc.prepareTestClient(t)
 		tc.testMetricsClient = testMetricsClient
 		tc.testCMClient = testCMClient
 
@@ -1447,7 +1613,7 @@ func TestConditionFailedGetScale(t *testing.T) {
 		},
 	}
 
-	_, _, _, testScaleClient := tc.prepareTestClient(t)
+	_, _, _, _, testScaleClient := tc.prepareTestClient(t)
 	tc.testScaleClient = testScaleClient
 
 	testScaleClient.PrependReactor("get", "replicationcontrollers", func(action core.Action) (handled bool, ret runtime.Object, err error) {
@@ -1474,7 +1640,7 @@ func TestConditionFailedUpdateScale(t *testing.T) {
 		}),
 	}
 
-	_, _, _, testScaleClient := tc.prepareTestClient(t)
+	_, _, _, _, testScaleClient := tc.prepareTestClient(t)
 	tc.testScaleClient = testScaleClient
 
 	testScaleClient.PrependReactor("update", "replicationcontrollers", func(action core.Action) (handled bool, ret runtime.Object, err error) {
@@ -1660,7 +1826,7 @@ func TestAvoidUncessaryUpdates(t *testing.T) {
 		reportedPodReadiness: []v1.ConditionStatus{v1.ConditionTrue, v1.ConditionFalse, v1.ConditionFalse},
 		useMetricsAPI:        true,
 	}
-	testClient, _, _, _ := tc.prepareTestClient(t)
+	testClient, _, _, _, _ := tc.prepareTestClient(t)
 	tc.testClient = testClient
 	var savedHPA *autoscalingv1.HorizontalPodAutoscaler
 	testClient.PrependReactor("list", "horizontalpodautoscalers", func(action core.Action) (handled bool, ret runtime.Object, err error) {

pkg/controller/podautoscaler/metrics/BUILD

@@ -29,6 +29,7 @@ go_library(
         "//vendor/k8s.io/metrics/pkg/apis/metrics/v1alpha1:go_default_library",
         "//vendor/k8s.io/metrics/pkg/client/clientset_generated/clientset/typed/metrics/v1beta1:go_default_library",
         "//vendor/k8s.io/metrics/pkg/client/custom_metrics:go_default_library",
+        "//vendor/k8s.io/metrics/pkg/client/external_metrics:go_default_library",
     ],
 )
 
@@ -56,10 +57,12 @@ go_test(
         "//vendor/k8s.io/client-go/testing:go_default_library",
         "//vendor/k8s.io/heapster/metrics/api/v1/types:go_default_library",
         "//vendor/k8s.io/metrics/pkg/apis/custom_metrics/v1beta1:go_default_library",
+        "//vendor/k8s.io/metrics/pkg/apis/external_metrics/v1beta1:go_default_library",
         "//vendor/k8s.io/metrics/pkg/apis/metrics/v1alpha1:go_default_library",
         "//vendor/k8s.io/metrics/pkg/apis/metrics/v1beta1:go_default_library",
         "//vendor/k8s.io/metrics/pkg/client/clientset_generated/clientset/fake:go_default_library",
         "//vendor/k8s.io/metrics/pkg/client/custom_metrics/fake:go_default_library",
+        "//vendor/k8s.io/metrics/pkg/client/external_metrics/fake:go_default_library",
     ],
 )
 

pkg/controller/podautoscaler/metrics/interfaces.go

@@ -42,4 +42,8 @@ type MetricsClient interface {
 	// GetObjectMetric gets the given metric (and an associated timestamp) for the given
 	// object in the given namespace
 	GetObjectMetric(metricName string, namespace string, objectRef *autoscaling.CrossVersionObjectReference) (int64, time.Time, error)
+
+	// GetExternalMetric gets all the values of a given external metric
+	// that match the specified selector.
+	GetExternalMetric(metricName string, namespace string, selector labels.Selector) ([]int64, time.Time, error)
 }

pkg/controller/podautoscaler/metrics/legacy_metrics_client.go

@@ -177,6 +177,10 @@ func (h *HeapsterMetricsClient) GetObjectMetric(metricName string, namespace str
 	return 0, time.Time{}, fmt.Errorf("object metrics are not yet supported")
 }
 
+func (h *HeapsterMetricsClient) GetExternalMetric(metricName, namespace string, selector labels.Selector) ([]int64, time.Time, error) {
+	return nil, time.Time{}, fmt.Errorf("external metrics aren't supported")
+}
+
 func collapseTimeSamples(metrics heapster.MetricResult, duration time.Duration) (int64, time.Time, bool) {
 	floatSum := float64(0)
 	intSum := int64(0)

pkg/controller/podautoscaler/metrics/rest_metrics_client.go

@@ -30,12 +30,14 @@ import (
 	customapi "k8s.io/metrics/pkg/apis/custom_metrics/v1beta1"
 	resourceclient "k8s.io/metrics/pkg/client/clientset_generated/clientset/typed/metrics/v1beta1"
 	customclient "k8s.io/metrics/pkg/client/custom_metrics"
+	externalclient "k8s.io/metrics/pkg/client/external_metrics"
 )
 
-func NewRESTMetricsClient(resourceClient resourceclient.PodMetricsesGetter, customClient customclient.CustomMetricsClient) MetricsClient {
+func NewRESTMetricsClient(resourceClient resourceclient.PodMetricsesGetter, customClient customclient.CustomMetricsClient, externalClient externalclient.ExternalMetricsClient) MetricsClient {
 	return &restMetricsClient{
 		&resourceMetricsClient{resourceClient},
 		&customMetricsClient{customClient},
+		&externalMetricsClient{externalClient},
 	}
 }
 
@@ -45,6 +47,7 @@ func NewRESTMetricsClient(resourceClient resourceclient.PodMetricsesGetter, cust
 type restMetricsClient struct {
 	*resourceMetricsClient
 	*customMetricsClient
+	*externalMetricsClient
 }
 
 // resourceMetricsClient implements the resource-metrics-related parts of MetricsClient,
@@ -139,3 +142,29 @@ func (c *customMetricsClient) GetObjectMetric(metricName string, namespace strin
 
 	return metricValue.Value.MilliValue(), metricValue.Timestamp.Time, nil
 }
+
+// externalMetricsClient implenets the external metrics related parts of MetricsClient,
+// using data from the external metrics API.
+type externalMetricsClient struct {
+	client externalclient.ExternalMetricsClient
+}
+
+// GetExternalMetric gets all the values of a given external metric
+// that match the specified selector.
+func (c *externalMetricsClient) GetExternalMetric(metricName, namespace string, selector labels.Selector) ([]int64, time.Time, error) {
+	metrics, err := c.client.NamespacedMetrics(namespace).List(metricName, selector)
+	if err != nil {
+		return []int64{}, time.Time{}, fmt.Errorf("unable to fetch metrics from external metrics API: %v", err)
+	}
+
+	if len(metrics.Items) == 0 {
+		return nil, time.Time{}, fmt.Errorf("no metrics returned from external metrics API")
+	}
+
+	res := make([]int64, 0)
+	for _, m := range metrics.Items {
+		res = append(res, m.Value.MilliValue())
+	}
+	timestamp := metrics.Items[0].Timestamp.Time
+	return res, timestamp, nil
+}

pkg/controller/podautoscaler/metrics/rest_metrics_client_test.go

@@ -32,9 +32,11 @@ import (
 	"k8s.io/kubernetes/pkg/api/legacyscheme"
 	_ "k8s.io/kubernetes/pkg/apis/extensions/install"
 	cmapi "k8s.io/metrics/pkg/apis/custom_metrics/v1beta1"
+	emapi "k8s.io/metrics/pkg/apis/external_metrics/v1beta1"
 	metricsapi "k8s.io/metrics/pkg/apis/metrics/v1beta1"
 	metricsfake "k8s.io/metrics/pkg/client/clientset_generated/clientset/fake"
 	cmfake "k8s.io/metrics/pkg/client/custom_metrics/fake"
+	emfake "k8s.io/metrics/pkg/client/external_metrics/fake"
 
 	"github.com/stretchr/testify/assert"
 )
@@ -53,9 +55,11 @@ type restClientTestCase struct {
 	selector            labels.Selector
 	resourceName        v1.ResourceName
 	metricName          string
+	metricSelector      *metav1.LabelSelector
+	metricLabelSelector labels.Selector
 }
 
-func (tc *restClientTestCase) prepareTestClient(t *testing.T) (*metricsfake.Clientset, *cmfake.FakeCustomMetricsClient) {
+func (tc *restClientTestCase) prepareTestClient(t *testing.T) (*metricsfake.Clientset, *cmfake.FakeCustomMetricsClient, *emfake.FakeExternalMetricsClient) {
 	namespace := "test-namespace"
 	tc.namespace = namespace
 	podNamePrefix := "test-pod"
@@ -64,9 +68,12 @@ func (tc *restClientTestCase) prepareTestClient(t *testing.T) (*metricsfake.Clie
 
 	// it's a resource test if we have a resource name
 	isResource := len(tc.resourceName) > 0
+	// it's an external test if we have a metric selector
+	isExternal := tc.metricSelector != nil
 
 	fakeMetricsClient := &metricsfake.Clientset{}
 	fakeCMClient := &cmfake.FakeCustomMetricsClient{}
+	fakeEMClient := &emfake.FakeExternalMetricsClient{}
 
 	if isResource {
 		fakeMetricsClient.AddReactor("list", "pods", func(action core.Action) (handled bool, ret runtime.Object, err error) {
@@ -99,6 +106,24 @@ func (tc *restClientTestCase) prepareTestClient(t *testing.T) (*metricsfake.Clie
 			}
 			return true, metrics, nil
 		})
+	} else if isExternal {
+		fakeEMClient.AddReactor("list", "*", func(action core.Action) (handled bool, ret runtime.Object, err error) {
+			listAction := action.(core.ListAction)
+			assert.Equal(t, tc.metricName, listAction.GetResource().Resource, "the metric requested should have matched the one specified.")
+			assert.Equal(t, tc.metricLabelSelector, listAction.GetListRestrictions().Labels, "the metric selector should have matched the one specified")
+
+			metrics := emapi.ExternalMetricValueList{}
+			for _, metricPoint := range tc.reportedMetricPoints {
+				timestamp := fixedTimestamp.Add(time.Duration(metricPoint.timestamp) * time.Minute)
+				metric := emapi.ExternalMetricValue{
+					Value:      *resource.NewMilliQuantity(int64(metricPoint.level), resource.DecimalSI),
+					Timestamp:  metav1.Time{Time: timestamp},
+					MetricName: tc.metricName,
+				}
+				metrics.Items = append(metrics.Items, metric)
+			}
+			return true, &metrics, nil
+		})
 	} else {
 		fakeCMClient.AddReactor("get", "*", func(action core.Action) (handled bool, ret runtime.Object, err error) {
 			getForAction := action.(cmfake.GetForAction)
@@ -162,13 +187,13 @@ func (tc *restClientTestCase) prepareTestClient(t *testing.T) (*metricsfake.Clie
 		})
 	}
 
-	return fakeMetricsClient, fakeCMClient
+	return fakeMetricsClient, fakeCMClient, fakeEMClient
 }
 
 func (tc *restClientTestCase) verifyResults(t *testing.T, metrics PodMetricsInfo, timestamp time.Time, err error) {
 	if tc.desiredError != nil {
 		assert.Error(t, err, "there should be an error retrieving the metrics")
-		assert.Contains(t, fmt.Sprintf("%v", err), fmt.Sprintf("%v", tc.desiredError), "the error message should be eas expected")
+		assert.Contains(t, fmt.Sprintf("%v", err), fmt.Sprintf("%v", tc.desiredError), "the error message should be as expected")
 		return
 	}
 	assert.NoError(t, err, "there should be no error retrieving the metrics")
@@ -181,12 +206,25 @@ func (tc *restClientTestCase) verifyResults(t *testing.T, metrics PodMetricsInfo
 }
 
 func (tc *restClientTestCase) runTest(t *testing.T) {
-	testMetricsClient, testCMClient := tc.prepareTestClient(t)
+	var err error
-	metricsClient := NewRESTMetricsClient(testMetricsClient.MetricsV1beta1(), testCMClient)
+	testMetricsClient, testCMClient, testEMClient := tc.prepareTestClient(t)
+	metricsClient := NewRESTMetricsClient(testMetricsClient.MetricsV1beta1(), testCMClient, testEMClient)
 	isResource := len(tc.resourceName) > 0
+	isExternal := tc.metricSelector != nil
 	if isResource {
 		info, timestamp, err := metricsClient.GetResourceMetric(v1.ResourceName(tc.resourceName), tc.namespace, tc.selector)
 		tc.verifyResults(t, info, timestamp, err)
+	} else if isExternal {
+		tc.metricLabelSelector, err = metav1.LabelSelectorAsSelector(tc.metricSelector)
+		if err != nil {
+			t.Errorf("invalid metric selector: %+v", tc.metricSelector)
+		}
+		val, timestamp, err := metricsClient.GetExternalMetric(tc.metricName, tc.namespace, tc.metricLabelSelector)
+		info := make(PodMetricsInfo, len(val))
+		for i, metricVal := range val {
+			info[fmt.Sprintf("%v-val-%v", tc.metricName, i)] = metricVal
+		}
+		tc.verifyResults(t, info, timestamp, err)
 	} else if tc.singleObject == nil {
 		info, timestamp, err := metricsClient.GetRawMetric(tc.metricName, tc.namespace, tc.selector)
 		tc.verifyResults(t, info, timestamp, err)
@@ -209,6 +247,19 @@ func TestRESTClientCPU(t *testing.T) {
 	tc.runTest(t)
 }
 
+func TestRESTClientExternal(t *testing.T) {
+	tc := restClientTestCase{
+		desiredMetricValues: PodMetricsInfo{
+			"external-val-0": 10000, "external-val-1": 20000, "external-val-2": 10000,
+		},
+		metricSelector:       &metav1.LabelSelector{MatchLabels: map[string]string{"label": "value"}},
+		metricName:           "external",
+		targetTimestamp:      1,
+		reportedMetricPoints: []metricPoint{{10000, 1}, {20000, 1}, {10000, 1}},
+	}
+	tc.runTest(t)
+}
+
 func TestRESTClientQPS(t *testing.T) {
 	tc := restClientTestCase{
 		desiredMetricValues: PodMetricsInfo{
@@ -248,6 +299,19 @@ func TestRESTClientQpsSumEqualZero(t *testing.T) {
 	tc.runTest(t)
 }
 
+func TestRESTClientExternalSumEqualZero(t *testing.T) {
+	tc := restClientTestCase{
+		desiredMetricValues: PodMetricsInfo{
+			"external-val-0": 0, "external-val-1": 0, "external-val-2": 0,
+		},
+		metricSelector:       &metav1.LabelSelector{MatchLabels: map[string]string{"label": "value"}},
+		metricName:           "external",
+		targetTimestamp:      0,
+		reportedMetricPoints: []metricPoint{{0, 0}, {0, 0}, {0, 0}},
+	}
+	tc.runTest(t)
+}
+
 func TestRESTClientQpsEmptyMetrics(t *testing.T) {
 	tc := restClientTestCase{
 		metricName:           "qps",
@@ -258,6 +322,17 @@ func TestRESTClientQpsEmptyMetrics(t *testing.T) {
 	tc.runTest(t)
 }
 
+func TestRESTClientExternalEmptyMetrics(t *testing.T) {
+	tc := restClientTestCase{
+		metricName:           "external",
+		metricSelector:       &metav1.LabelSelector{MatchLabels: map[string]string{"label": "value"}},
+		desiredError:         fmt.Errorf("no metrics returned from external metrics API"),
+		reportedMetricPoints: []metricPoint{},
+	}
+
+	tc.runTest(t)
+}
+
 func TestRESTClientCPUEmptyMetrics(t *testing.T) {
 	tc := restClientTestCase{
 		resourceName:         v1.ResourceCPU,

pkg/controller/podautoscaler/replica_calculator.go

@@ -278,6 +278,33 @@ func (c *ReplicaCalculator) GetObjectMetricReplicas(currentReplicas int32, targe
 		return 0, 0, time.Time{}, fmt.Errorf("unable to get metric %s: %v on %s %s/%s", metricName, objectRef.Kind, namespace, objectRef.Name, err)
 	}
 
+	usageRatio := float64(utilization) / float64(targetUtilization)
+	if math.Abs(1.0-usageRatio) <= c.tolerance {
+		// return the current replicas if the change would be too small
+		return currentReplicas, utilization, timestamp, nil
+	}
+	replicaCount = int32(math.Ceil(usageRatio * float64(currentReplicas)))
+
+	return replicaCount, utilization, timestamp, nil
+}
+
+// GetExternalMetricReplicas calculates the desired replica count based on a
+// target metric value (as a milli-value) for the external metric in the given
+// namespace, and the current replica count.
+func (c *ReplicaCalculator) GetExternalMetricReplicas(currentReplicas int32, targetUtilization int64, metricName, namespace string, selector *metav1.LabelSelector) (replicaCount int32, utilization int64, timestamp time.Time, err error) {
+	labelSelector, err := metav1.LabelSelectorAsSelector(selector)
+	if err != nil {
+		return 0, 0, time.Time{}, err
+	}
+	metrics, timestamp, err := c.metricsClient.GetExternalMetric(metricName, namespace, labelSelector)
+	if err != nil {
+		return 0, 0, time.Time{}, fmt.Errorf("unable to get external metric %s/%s/%+v: %s", namespace, metricName, selector, err)
+	}
+	utilization = 0
+	for _, val := range metrics {
+		utilization = utilization + val
+	}
+
 	usageRatio := float64(utilization) / float64(targetUtilization)
 	if math.Abs(1.0-usageRatio) <= c.tolerance {
 		// return the current replicas if the change would be too small
@@ -286,3 +313,30 @@ func (c *ReplicaCalculator) GetObjectMetricReplicas(currentReplicas int32, targe
 
 	return int32(math.Ceil(usageRatio * float64(currentReplicas))), utilization, timestamp, nil
 }
+
+// GetExternalPerPodMetricReplicas calculates the desired replica count based on a
+// target metric value per pod (as a milli-value) for the external metric in the
+// given namespace, and the current replica count.
+func (c *ReplicaCalculator) GetExternalPerPodMetricReplicas(currentReplicas int32, targetUtilizationPerPod int64, metricName, namespace string, selector *metav1.LabelSelector) (replicaCount int32, utilization int64, timestamp time.Time, err error) {
+	labelSelector, err := metav1.LabelSelectorAsSelector(selector)
+	if err != nil {
+		return 0, 0, time.Time{}, err
+	}
+	metrics, timestamp, err := c.metricsClient.GetExternalMetric(metricName, namespace, labelSelector)
+	if err != nil {
+		return 0, 0, time.Time{}, fmt.Errorf("unable to get external metric %s/%s/%+v: %s", namespace, metricName, selector, err)
+	}
+	utilization = 0
+	for _, val := range metrics {
+		utilization = utilization + val
+	}
+
+	replicaCount = currentReplicas
+	usageRatio := float64(utilization) / (float64(targetUtilizationPerPod) * float64(replicaCount))
+	if math.Abs(1.0-usageRatio) > c.tolerance {
+		// update number of replicas if the change is large enough
+		replicaCount = int32(math.Ceil(float64(utilization) / float64(targetUtilizationPerPod)))
+	}
+	utilization = int64(math.Ceil(float64(utilization) / float64(currentReplicas)))
+	return replicaCount, utilization, timestamp, nil
+}

pkg/controller/podautoscaler/replica_calculator_test.go

@@ -33,9 +33,11 @@ import (
 	"k8s.io/kubernetes/pkg/api/legacyscheme"
 	"k8s.io/kubernetes/pkg/controller/podautoscaler/metrics"
 	cmapi "k8s.io/metrics/pkg/apis/custom_metrics/v1beta1"
+	emapi "k8s.io/metrics/pkg/apis/external_metrics/v1beta1"
 	metricsapi "k8s.io/metrics/pkg/apis/metrics/v1beta1"
 	metricsfake "k8s.io/metrics/pkg/client/clientset_generated/clientset/fake"
 	cmfake "k8s.io/metrics/pkg/client/custom_metrics/fake"
+	emfake "k8s.io/metrics/pkg/client/external_metrics/fake"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -57,8 +59,10 @@ type metricInfo struct {
 	name         string
 	levels       []int64
 	singleObject *autoscalingv2.CrossVersionObjectReference
+	selector     *metav1.LabelSelector
 
 	targetUtilization       int64
+	perPodTargetUtilization int64
 	expectedUtilization     int64
 }
 
@@ -81,7 +85,7 @@ const (
 	numContainersPerPod = 2
 )
 
-func (tc *replicaCalcTestCase) prepareTestClient(t *testing.T) (*fake.Clientset, *metricsfake.Clientset, *cmfake.FakeCustomMetricsClient) {
+func (tc *replicaCalcTestCase) prepareTestClient(t *testing.T) (*fake.Clientset, *metricsfake.Clientset, *cmfake.FakeCustomMetricsClient, *emfake.FakeExternalMetricsClient) {
 
 	fakeClient := &fake.Clientset{}
 	fakeClient.AddReactor("list", "pods", func(action core.Action) (handled bool, ret runtime.Object, err error) {
@@ -236,12 +240,45 @@ func (tc *replicaCalcTestCase) prepareTestClient(t *testing.T) (*fake.Clientset,
 		return true, metrics, nil
 	})
 
-	return fakeClient, fakeMetricsClient, fakeCMClient
+	fakeEMClient := &emfake.FakeExternalMetricsClient{}
+	fakeEMClient.AddReactor("list", "*", func(action core.Action) (handled bool, ret runtime.Object, err error) {
+		listAction, wasList := action.(core.ListAction)
+		if !wasList {
+			return true, nil, fmt.Errorf("expected a list-for action, got %v instead", action)
+		}
+
+		if tc.metric == nil {
+			return true, nil, fmt.Errorf("no external metrics specified in test client")
+		}
+
+		assert.Equal(t, tc.metric.name, listAction.GetResource().Resource, "the metric requested should have matched the one specified")
+
+		selector, err := metav1.LabelSelectorAsSelector(tc.metric.selector)
+		if err != nil {
+			return true, nil, fmt.Errorf("failed to convert label selector specified in test client")
+		}
+		assert.Equal(t, selector, listAction.GetListRestrictions().Labels, "the metric selector should have matched the one specified")
+
+		metrics := emapi.ExternalMetricValueList{}
+
+		for _, level := range tc.metric.levels {
+			metric := emapi.ExternalMetricValue{
+				Timestamp:  metav1.Time{Time: tc.timestamp},
+				MetricName: tc.metric.name,
+				Value:      *resource.NewMilliQuantity(level, resource.DecimalSI),
+			}
+			metrics.Items = append(metrics.Items, metric)
+		}
+
+		return true, &metrics, nil
+	})
+
+	return fakeClient, fakeMetricsClient, fakeCMClient, fakeEMClient
 }
 
 func (tc *replicaCalcTestCase) runTest(t *testing.T) {
-	testClient, testMetricsClient, testCMClient := tc.prepareTestClient(t)
+	testClient, testMetricsClient, testCMClient, testEMClient := tc.prepareTestClient(t)
-	metricsClient := metrics.NewRESTMetricsClient(testMetricsClient.MetricsV1beta1(), testCMClient)
+	metricsClient := metrics.NewRESTMetricsClient(testMetricsClient.MetricsV1beta1(), testCMClient, testEMClient)
 
 	replicaCalc := &ReplicaCalculator{
 		metricsClient: metricsClient,
@@ -277,6 +314,12 @@ func (tc *replicaCalcTestCase) runTest(t *testing.T) {
 		var err error
 		if tc.metric.singleObject != nil {
 			outReplicas, outUtilization, outTimestamp, err = replicaCalc.GetObjectMetricReplicas(tc.currentReplicas, tc.metric.targetUtilization, tc.metric.name, testNamespace, tc.metric.singleObject)
+		} else if tc.metric.selector != nil {
+			if tc.metric.targetUtilization > 0 {
+				outReplicas, outUtilization, outTimestamp, err = replicaCalc.GetExternalMetricReplicas(tc.currentReplicas, tc.metric.targetUtilization, tc.metric.name, testNamespace, tc.metric.selector)
+			} else if tc.metric.perPodTargetUtilization > 0 {
+				outReplicas, outUtilization, outTimestamp, err = replicaCalc.GetExternalPerPodMetricReplicas(tc.currentReplicas, tc.metric.perPodTargetUtilization, tc.metric.name, testNamespace, tc.metric.selector)
+			}
 		} else {
 			outReplicas, outUtilization, outTimestamp, err = replicaCalc.GetMetricReplicas(tc.currentReplicas, tc.metric.targetUtilization, tc.metric.name, testNamespace, selector)
 		}
@@ -425,6 +468,50 @@ func TestReplicaCalcScaleUpCMObject(t *testing.T) {
 	tc.runTest(t)
 }
 
+func TestReplicaCalcScaleUpCMExternal(t *testing.T) {
+	tc := replicaCalcTestCase{
+		currentReplicas:  1,
+		expectedReplicas: 2,
+		metric: &metricInfo{
+			name:                "qps",
+			levels:              []int64{8600},
+			targetUtilization:   4400,
+			expectedUtilization: 8600,
+			selector:            &metav1.LabelSelector{MatchLabels: map[string]string{"label": "value"}},
+		},
+	}
+	tc.runTest(t)
+}
+
+func TestReplicaCalcScaleUpCMExternalNoLabels(t *testing.T) {
+	tc := replicaCalcTestCase{
+		currentReplicas:  1,
+		expectedReplicas: 2,
+		metric: &metricInfo{
+			name:                "qps",
+			levels:              []int64{8600},
+			targetUtilization:   4400,
+			expectedUtilization: 8600,
+		},
+	}
+	tc.runTest(t)
+}
+
+func TestReplicaCalcScaleUpPerPodCMExternal(t *testing.T) {
+	tc := replicaCalcTestCase{
+		currentReplicas:  3,
+		expectedReplicas: 4,
+		metric: &metricInfo{
+			name:                    "qps",
+			levels:                  []int64{8600},
+			perPodTargetUtilization: 2150,
+			expectedUtilization:     2867,
+			selector:                &metav1.LabelSelector{MatchLabels: map[string]string{"label": "value"}},
+		},
+	}
+	tc.runTest(t)
+}
+
 func TestReplicaCalcScaleDown(t *testing.T) {
 	tc := replicaCalcTestCase{
 		currentReplicas:  5,
@@ -475,6 +562,36 @@ func TestReplicaCalcScaleDownCMObject(t *testing.T) {
 	tc.runTest(t)
 }
 
+func TestReplicaCalcScaleDownCMExternal(t *testing.T) {
+	tc := replicaCalcTestCase{
+		currentReplicas:  5,
+		expectedReplicas: 3,
+		metric: &metricInfo{
+			name:                "qps",
+			levels:              []int64{8600},
+			targetUtilization:   14334,
+			expectedUtilization: 8600,
+			selector:            &metav1.LabelSelector{MatchLabels: map[string]string{"label": "value"}},
+		},
+	}
+	tc.runTest(t)
+}
+
+func TestReplicaCalcScaleDownPerPodCMExternal(t *testing.T) {
+	tc := replicaCalcTestCase{
+		currentReplicas:  5,
+		expectedReplicas: 3,
+		metric: &metricInfo{
+			name:                    "qps",
+			levels:                  []int64{8600},
+			perPodTargetUtilization: 2867,
+			expectedUtilization:     1720,
+			selector:                &metav1.LabelSelector{MatchLabels: map[string]string{"label": "value"}},
+		},
+	}
+	tc.runTest(t)
+}
+
 func TestReplicaCalcScaleDownIgnoresUnreadyPods(t *testing.T) {
 	tc := replicaCalcTestCase{
 		currentReplicas:  5,
@@ -543,6 +660,36 @@ func TestReplicaCalcToleranceCMObject(t *testing.T) {
 	tc.runTest(t)
 }
 
+func TestReplicaCalcToleranceCMExternal(t *testing.T) {
+	tc := replicaCalcTestCase{
+		currentReplicas:  3,
+		expectedReplicas: 3,
+		metric: &metricInfo{
+			name:                "qps",
+			levels:              []int64{8600},
+			targetUtilization:   8888,
+			expectedUtilization: 8600,
+			selector:            &metav1.LabelSelector{MatchLabels: map[string]string{"label": "value"}},
+		},
+	}
+	tc.runTest(t)
+}
+
+func TestReplicaCalcTolerancePerPodCMExternal(t *testing.T) {
+	tc := replicaCalcTestCase{
+		currentReplicas:  3,
+		expectedReplicas: 3,
+		metric: &metricInfo{
+			name:                    "qps",
+			levels:                  []int64{8600},
+			perPodTargetUtilization: 2900,
+			expectedUtilization:     2867,
+			selector:                &metav1.LabelSelector{MatchLabels: map[string]string{"label": "value"}},
+		},
+	}
+	tc.runTest(t)
+}
+
 func TestReplicaCalcSuperfluousMetrics(t *testing.T) {
 	tc := replicaCalcTestCase{
 		currentReplicas:  4,