HPA: Consider unready pods and missing metrics

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

@@ -427,7 +427,8 @@ func StartControllers(s *options.CMServer, kubeconfig *restclient.Config, rootCl
 				metrics.DefaultHeapsterService,
 				metrics.DefaultHeapsterPort,
 			)
-			go podautoscaler.NewHorizontalController(hpaClient.Core(), hpaClient.Extensions(), hpaClient.Autoscaling(), metricsClient, s.HorizontalPodAutoscalerSyncPeriod.Duration).
+			replicaCalc := podautoscaler.NewReplicaCalculator(metricsClient, hpaClient.Core())
+			go podautoscaler.NewHorizontalController(hpaClient.Core(), hpaClient.Extensions(), hpaClient.Autoscaling(), replicaCalc, s.HorizontalPodAutoscalerSyncPeriod.Duration).
 				Run(wait.NeverStop)
 			time.Sleep(wait.Jitter(s.ControllerStartInterval.Duration, ControllerStartJitter))
 		}

pkg/controller/podautoscaler/BUILD

@@ -15,6 +15,7 @@ go_library(
     srcs = [
         "doc.go",
         "horizontal.go",
+        "replica_calculator.go",
     ],
     tags = ["automanaged"],
     deps = [
@@ -29,8 +30,10 @@ go_library(
         "//pkg/client/clientset_generated/internalclientset/typed/extensions/internalversion:go_default_library",
         "//pkg/client/record:go_default_library",
         "//pkg/controller/podautoscaler/metrics:go_default_library",
+        "//pkg/labels:go_default_library",
         "//pkg/runtime:go_default_library",
         "//pkg/util/runtime:go_default_library",
+        "//pkg/util/sets:go_default_library",
         "//pkg/watch:go_default_library",
         "//vendor:github.com/golang/glog",
     ],
@@ -38,7 +41,10 @@ go_library(
 
 go_test(
     name = "go_default_test",
-    srcs = ["horizontal_test.go"],
+    srcs = [
+        "horizontal_test.go",
+        "replica_calculator_test.go",
+    ],
     library = "go_default_library",
     tags = ["automanaged"],
     deps = [
@@ -58,6 +64,7 @@ go_test(
         "//pkg/runtime:go_default_library",
         "//pkg/watch:go_default_library",
         "//vendor:github.com/stretchr/testify/assert",
+        "//vendor:github.com/stretchr/testify/require",
         "//vendor:k8s.io/heapster/metrics/api/v1/types",
         "//vendor:k8s.io/heapster/metrics/apis/metrics/v1alpha1",
     ],

pkg/controller/podautoscaler/horizontal.go

@@ -33,7 +33,6 @@ import (
 	unversionedcore "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/typed/core/internalversion"
 	unversionedextensions "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/typed/extensions/internalversion"
 	"k8s.io/kubernetes/pkg/client/record"
-	"k8s.io/kubernetes/pkg/controller/podautoscaler/metrics"
 	"k8s.io/kubernetes/pkg/runtime"
 	utilruntime "k8s.io/kubernetes/pkg/util/runtime"
 	"k8s.io/kubernetes/pkg/watch"
@@ -61,7 +60,7 @@ type HorizontalController struct {
 	scaleNamespacer unversionedextensions.ScalesGetter
 	hpaNamespacer   unversionedautoscaling.HorizontalPodAutoscalersGetter
 
-	metricsClient metrics.MetricsClient
+	replicaCalc   *ReplicaCalculator
 	eventRecorder record.EventRecorder
 
 	// A store of HPA objects, populated by the controller.
@@ -110,13 +109,13 @@ func newInformer(controller *HorizontalController, resyncPeriod time.Duration) (
 	)
 }
 
-func NewHorizontalController(evtNamespacer unversionedcore.EventsGetter, scaleNamespacer unversionedextensions.ScalesGetter, hpaNamespacer unversionedautoscaling.HorizontalPodAutoscalersGetter, metricsClient metrics.MetricsClient, resyncPeriod time.Duration) *HorizontalController {
+func NewHorizontalController(evtNamespacer unversionedcore.EventsGetter, scaleNamespacer unversionedextensions.ScalesGetter, hpaNamespacer unversionedautoscaling.HorizontalPodAutoscalersGetter, replicaCalc *ReplicaCalculator, resyncPeriod time.Duration) *HorizontalController {
 	broadcaster := record.NewBroadcaster()
 	broadcaster.StartRecordingToSink(&unversionedcore.EventSinkImpl{Interface: evtNamespacer.Events("")})
 	recorder := broadcaster.NewRecorder(api.EventSource{Component: "horizontal-pod-autoscaler"})
 
 	controller := &HorizontalController{
-		metricsClient:   metricsClient,
+		replicaCalc:     replicaCalc,
 		eventRecorder:   recorder,
 		scaleNamespacer: scaleNamespacer,
 		hpaNamespacer:   hpaNamespacer,
@@ -164,9 +163,8 @@ func (a *HorizontalController) computeReplicasForCPUUtilization(hpa *autoscaling
 		a.eventRecorder.Event(hpa, api.EventTypeWarning, "InvalidSelector", errMsg)
 		return 0, nil, time.Time{}, fmt.Errorf(errMsg)
 	}
-	currentUtilization, numRunningPods, timestamp, err := a.metricsClient.GetCPUUtilization(hpa.Namespace, selector)
 
-	// TODO: what to do on partial errors (like metrics obtained for 75% of pods).
+	desiredReplicas, utilization, timestamp, err := a.replicaCalc.GetResourceReplicas(currentReplicas, targetUtilization, api.ResourceCPU, hpa.Namespace, selector)
 	if err != nil {
 		lastScaleTime := getLastScaleTime(hpa)
 		if time.Now().After(lastScaleTime.Add(upscaleForbiddenWindow)) {
@@ -178,20 +176,13 @@ func (a *HorizontalController) computeReplicasForCPUUtilization(hpa *autoscaling
 		return 0, nil, time.Time{}, fmt.Errorf("failed to get CPU utilization: %v", err)
 	}
 
-	utilization := int32(*currentUtilization)
-
-	usageRatio := float64(utilization) / float64(targetUtilization)
-	if math.Abs(1.0-usageRatio) <= tolerance {
-		return currentReplicas, &utilization, timestamp, nil
+	if desiredReplicas != currentReplicas {
+		a.eventRecorder.Eventf(hpa, api.EventTypeNormal, "DesiredReplicasComputed",
+			"Computed the desired num of replicas: %d (avgCPUutil: %d, current replicas: %d)",
+			desiredReplicas, utilization, scale.Status.Replicas)
 	}
 
-	desiredReplicas := math.Ceil(usageRatio * float64(numRunningPods))
-
-	a.eventRecorder.Eventf(hpa, api.EventTypeNormal, "DesiredReplicasComputed",
-		"Computed the desired num of replicas: %d, on a base of %d report(s) (avgCPUutil: %d, current replicas: %d)",
-		int32(desiredReplicas), numRunningPods, utilization, scale.Status.Replicas)
-
-	return int32(desiredReplicas), &utilization, timestamp, nil
+	return desiredReplicas, &utilization, timestamp, nil
 }
 
 // computeReplicasForCustomMetrics computes the desired number of replicas based on the CustomMetrics passed in cmAnnotation
@@ -233,8 +224,8 @@ func (a *HorizontalController) computeReplicasForCustomMetrics(hpa *autoscaling.
 			a.eventRecorder.Event(hpa, api.EventTypeWarning, "InvalidSelector", errMsg)
 			return 0, "", "", time.Time{}, fmt.Errorf("couldn't convert selector string to a corresponding selector object: %v", err)
 		}
-		value, currentTimestamp, err := a.metricsClient.GetCustomMetric(customMetricTarget.Name, hpa.Namespace, selector)
-		// TODO: what to do on partial errors (like metrics obtained for 75% of pods).
+		floatTarget := float64(customMetricTarget.TargetValue.MilliValue()) / 1000.0
+		replicaCountProposal, utilizationProposal, timestampProposal, err := a.replicaCalc.GetMetricReplicas(currentReplicas, floatTarget, fmt.Sprintf("custom/%s", customMetricTarget.Name), hpa.Namespace, selector)
 		if err != nil {
 			lastScaleTime := getLastScaleTime(hpa)
 			if time.Now().After(lastScaleTime.Add(upscaleForbiddenWindow)) {
@@ -245,21 +236,13 @@ func (a *HorizontalController) computeReplicasForCustomMetrics(hpa *autoscaling.
 
 			return 0, "", "", time.Time{}, fmt.Errorf("failed to get custom metric value: %v", err)
 		}
-		floatTarget := float64(customMetricTarget.TargetValue.MilliValue()) / 1000.0
-		usageRatio := *value / floatTarget
 
-		replicaCountProposal := int32(0)
-		if math.Abs(1.0-usageRatio) > tolerance {
-			replicaCountProposal = int32(math.Ceil(usageRatio * float64(currentReplicas)))
-		} else {
-			replicaCountProposal = currentReplicas
-		}
 		if replicaCountProposal > replicas {
-			timestamp = currentTimestamp
+			timestamp = timestampProposal
 			replicas = replicaCountProposal
 			metric = fmt.Sprintf("Custom metric %s", customMetricTarget.Name)
 		}
-		quantity, err := resource.ParseQuantity(fmt.Sprintf("%.3f", *value))
+		quantity, err := resource.ParseQuantity(fmt.Sprintf("%.3f", utilizationProposal))
 		if err != nil {
 			return 0, "", "", time.Time{}, fmt.Errorf("failed to set custom metric value: %v", err)
 		}

pkg/controller/podautoscaler/metrics/BUILD

@@ -12,14 +12,17 @@ load(
 
 go_library(
     name = "go_default_library",
-    srcs = ["metrics_client.go"],
+    srcs = [
+        "metrics_client.go",
+        "utilization.go",
+    ],
     tags = ["automanaged"],
     deps = [
         "//pkg/api:go_default_library",
         "//pkg/api/v1:go_default_library",
         "//pkg/client/clientset_generated/internalclientset:go_default_library",
+        "//pkg/client/clientset_generated/internalclientset/typed/core/internalversion:go_default_library",
         "//pkg/labels:go_default_library",
-        "//pkg/util/sets:go_default_library",
         "//vendor:github.com/golang/glog",
         "//vendor:k8s.io/heapster/metrics/api/v1/types",
         "//vendor:k8s.io/heapster/metrics/apis/metrics/v1alpha1",

pkg/controller/podautoscaler/metrics/utilization.go

+/*
+Copyright 2015 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package metrics
+
+// GetResourceUtilizationRatio takes in a set of metrics, a set of matching requests,
+// and a target utilization percentage, and calcuates the the ratio of
+// desired to actual utilization (returning that and the actual utilization)
+func GetResourceUtilizationRatio(metrics PodResourceInfo, requests map[string]int64, targetUtilization int32) (float64, int32, error) {
+	metricsTotal := int64(0)
+	requestsTotal := int64(0)
+
+	for podName, metricValue := range metrics {
+		request, hasRequest := requests[podName]
+		if !hasRequest {
+			// we check for missing requests elsewhere, so assuming missing requests == extraneous metrics
+			continue
+		}
+
+		metricsTotal += metricValue
+		requestsTotal += request
+	}
+
+	currentUtilization := int32((metricsTotal * 100) / requestsTotal)
+
+	return float64(currentUtilization) / float64(targetUtilization), currentUtilization, nil
+}
+
+// GetMetricUtilizationRatio takes in a set of metrics and a target utilization value,
+// and calcuates the ratio of desired to actual utilization
+// (returning that and the actual utilization)
+func GetMetricUtilizationRatio(metrics PodMetricsInfo, targetUtilization float64) (float64, float64) {
+	metricsTotal := float64(0)
+	for _, metricValue := range metrics {
+		metricsTotal += metricValue
+	}
+
+	currentUtilization := metricsTotal / float64(len(metrics))
+
+	return currentUtilization / targetUtilization, currentUtilization
+}

pkg/controller/podautoscaler/replica_calculator.go

+/*
+Copyright 2016 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package podautoscaler
+
+import (
+	"fmt"
+	"math"
+	"time"
+
+	"k8s.io/kubernetes/pkg/api"
+	unversionedcore "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/typed/core/internalversion"
+	metricsclient "k8s.io/kubernetes/pkg/controller/podautoscaler/metrics"
+	"k8s.io/kubernetes/pkg/labels"
+	"k8s.io/kubernetes/pkg/util/sets"
+)
+
+type ReplicaCalculator struct {
+	metricsClient metricsclient.MetricsClient
+	podsGetter    unversionedcore.PodsGetter
+}
+
+func NewReplicaCalculator(metricsClient metricsclient.MetricsClient, podsGetter unversionedcore.PodsGetter) *ReplicaCalculator {
+	return &ReplicaCalculator{
+		metricsClient: metricsClient,
+		podsGetter:    podsGetter,
+	}
+}
+
+// GetResourceReplicas calculates the desired replica count based on a target resource utilization percentage
+// of the given resource for pods matching the given selector in the given namespace, and the current replica count
+func (c *ReplicaCalculator) GetResourceReplicas(currentReplicas int32, targetUtilization int32, resource api.ResourceName, namespace string, selector labels.Selector) (replicaCount int32, utilization int32, timestamp time.Time, err error) {
+	metrics, timestamp, err := c.metricsClient.GetResourceMetric(resource, namespace, selector)
+	if err != nil {
+		return 0, 0, time.Time{}, fmt.Errorf("unable to get metrics for resource %s: %v", resource, err)
+	}
+
+	podList, err := c.podsGetter.Pods(namespace).List(api.ListOptions{LabelSelector: selector})
+	if err != nil {
+		return 0, 0, time.Time{}, fmt.Errorf("unable to get pods while calculating replica count: %v", err)
+	}
+
+	if len(podList.Items) == 0 {
+		return 0, 0, time.Time{}, fmt.Errorf("no pods returned by selector while calculating replica count")
+	}
+
+	requests := make(map[string]int64, len(podList.Items))
+	readyPodCount := 0
+	unreadyPods := sets.NewString()
+	missingPods := sets.NewString()
+
+	for _, pod := range podList.Items {
+		podSum := int64(0)
+		for _, container := range pod.Spec.Containers {
+			if containerRequest, ok := container.Resources.Requests[resource]; ok {
+				podSum += containerRequest.MilliValue()
+			} else {
+				return 0, 0, time.Time{}, fmt.Errorf("missing request for %s on container %s in pod %s/%s", resource, container.Name, namespace, pod.Name)
+			}
+		}
+
+		requests[pod.Name] = podSum
+
+		if pod.Status.Phase != api.PodRunning || !api.IsPodReady(&pod) {
+			// save this pod name for later, but pretend it doesn't exist for now
+			unreadyPods.Insert(pod.Name)
+			delete(metrics, pod.Name)
+			continue
+		}
+
+		if _, found := metrics[pod.Name]; !found {
+			// save this pod name for later, but pretend it doesn't exist for now
+			missingPods.Insert(pod.Name)
+			continue
+		}
+
+		readyPodCount++
+	}
+
+	if len(metrics) == 0 {
+		return 0, 0, time.Time{}, fmt.Errorf("did not receive metrics for any ready pods")
+	}
+
+	usageRatio, utilization, err := metricsclient.GetResourceUtilizationRatio(metrics, requests, targetUtilization)
+	if err != nil {
+		return 0, 0, time.Time{}, err
+	}
+
+	rebalanceUnready := len(unreadyPods) > 0 && usageRatio > 1.0
+	if !rebalanceUnready && len(missingPods) == 0 {
+		if math.Abs(1.0-usageRatio) <= tolerance {
+			// return the current replicas if the change would be too small
+			return currentReplicas, utilization, timestamp, nil
+		}
+
+		// if we don't have any unready or missing pods, we can calculate the new replica count now
+		return int32(math.Ceil(usageRatio * float64(readyPodCount))), utilization, timestamp, nil
+	}
+
+	if len(missingPods) > 0 {
+		if usageRatio < 1.0 {
+			// on a scale-down, treat missing pods as using 100% of the resource request
+			for podName := range missingPods {
+				metrics[podName] = requests[podName]
+			}
+		} else {
+			// on a scale-up, treat missing pods as using 0% of the resource request
+			for podName := range missingPods {
+				metrics[podName] = 0
+			}
+		}
+	}
+
+	if rebalanceUnready {
+		// on a scale-up, treat unready pods as using 0% of the resource request
+		for podName := range unreadyPods {
+			metrics[podName] = 0
+		}
+	}
+
+	// re-run the utilization calculation with our new numbers
+	newUsageRatio, _, err := metricsclient.GetResourceUtilizationRatio(metrics, requests, targetUtilization)
+	if err != nil {
+		return 0, utilization, time.Time{}, err
+	}
+
+	if math.Abs(1.0-newUsageRatio) <= tolerance || (usageRatio < 1.0 && newUsageRatio > 1.0) || (usageRatio > 1.0 && newUsageRatio < 1.0) {
+		// return the current replicas if the change would be too small,
+		// or if the new usage ratio would cause a change in scale direction
+		return currentReplicas, utilization, timestamp, nil
+	}
+
+	// return the result, where the number of replicas considered is
+	// however many replicas factored into our calculation
+	return int32(math.Ceil(newUsageRatio * float64(len(metrics)))), utilization, timestamp, nil
+}
+
+// GetMetricReplicas calculates the desired replica count based on a target resource utilization percentage
+// of the given resource for pods matching the given selector in the given namespace, and the current replica count
+func (c *ReplicaCalculator) GetMetricReplicas(currentReplicas int32, targetUtilization float64, metricName string, namespace string, selector labels.Selector) (replicaCount int32, utilization float64, timestamp time.Time, err error) {
+	metrics, timestamp, err := c.metricsClient.GetRawMetric(metricName, namespace, selector)
+	if err != nil {
+		return 0, 0, time.Time{}, fmt.Errorf("unable to get metric  %s: %v", metricName, err)
+	}
+
+	podList, err := c.podsGetter.Pods(namespace).List(api.ListOptions{LabelSelector: selector})
+	if err != nil {
+		return 0, 0, time.Time{}, fmt.Errorf("unable to get pods while calculating replica count: %v", err)
+	}
+
+	if len(podList.Items) == 0 {
+		return 0, 0, time.Time{}, fmt.Errorf("no pods returned by selector while calculating replica count")
+	}
+
+	readyPodCount := 0
+	unreadyPods := sets.NewString()
+	missingPods := sets.NewString()
+
+	for _, pod := range podList.Items {
+		if pod.Status.Phase != api.PodRunning || !api.IsPodReady(&pod) {
+			// save this pod name for later, but pretend it doesn't exist for now
+			unreadyPods.Insert(pod.Name)
+			delete(metrics, pod.Name)
+			continue
+		}
+
+		if _, found := metrics[pod.Name]; !found {
+			// save this pod name for later, but pretend it doesn't exist for now
+			missingPods.Insert(pod.Name)
+			continue
+		}
+
+		readyPodCount++
+	}
+
+	if len(metrics) == 0 {
+		return 0, 0, time.Time{}, fmt.Errorf("did not recieve metrics for any ready pods")
+	}
+
+	usageRatio, utilization := metricsclient.GetMetricUtilizationRatio(metrics, targetUtilization)
+	if err != nil {
+		return 0, 0, time.Time{}, err
+	}
+
+	rebalanceUnready := len(unreadyPods) > 0 && usageRatio > 1.0
+
+	if !rebalanceUnready && len(missingPods) == 0 {
+		if math.Abs(1.0-usageRatio) <= tolerance {
+			// return the current replicas if the change would be too small
+			return currentReplicas, utilization, timestamp, nil
+		}
+
+		// if we don't have any unready or missing pods, we can calculate the new replica count now
+		return int32(math.Ceil(usageRatio * float64(readyPodCount))), utilization, timestamp, nil
+	}
+
+	if len(missingPods) > 0 {
+		if usageRatio < 1.0 {
+			// on a scale-down, treat missing pods as using 100% of the resource request
+			for podName := range missingPods {
+				metrics[podName] = targetUtilization
+			}
+		} else {
+			// on a scale-up, treat missing pods as using 0% of the resource request
+			for podName := range missingPods {
+				metrics[podName] = 0
+			}
+		}
+	}
+
+	if rebalanceUnready {
+		// on a scale-up, treat unready pods as using 0% of the resource request
+		for podName := range unreadyPods {
+			metrics[podName] = 0
+		}
+	}
+
+	// re-run the utilization calculation with our new numbers
+	newUsageRatio, _ := metricsclient.GetMetricUtilizationRatio(metrics, targetUtilization)
+	if err != nil {
+		return 0, utilization, time.Time{}, err
+	}
+
+	if math.Abs(1.0-newUsageRatio) <= tolerance || (usageRatio < 1.0 && newUsageRatio > 1.0) || (usageRatio > 1.0 && newUsageRatio < 1.0) {
+		// return the current replicas if the change would be too small,
+		// or if the new usage ratio would cause a change in scale direction
+		return currentReplicas, utilization, timestamp, nil
+	}
+
+	// return the result, where the number of replicas considered is
+	// however many replicas factored into our calculation
+	return int32(math.Ceil(newUsageRatio * float64(len(metrics)))), utilization, timestamp, nil
+}