【kubernetes/k8s源码分析】kubectl-controller-manager之HPA源码分析

àì夳堔傛蜴生んèń 2022-04-12 12:15 221阅读 0赞

*本文基于kubernetes版本：v1.12.1*

# HPA介绍 #

[https://kubernetes.io/docs/tasks/run-application/horizontal-pod-autoscale/][https_kubernetes.io_docs_tasks_run-application_horizontal-pod-autoscale]

The Horizontal Pod Autoscaler automatically scales the number of pods in a replication controller, deployment or replica set based on observed CPU utilization (or, with [custom metrics][] support, on some other application-provided metrics)

Horizontal Pod Autoscaling（HPA）是Kubernetes中实现POD水平自动伸缩的功能，根据CPU使用率或应用自定义`metrics`自动扩展Pod数量(支持 replication controller、deployment 和 replica set)

HPA是kubernetes中自动水平扩展模块，基于用户设定和获取到的指标(CPU,Memory,自定义metrics)，对Pod进行伸缩(不是直接操作pod)

# 原理： #

##   How does the Horizontal Pod Autoscaler work? ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob25nbGluemhhbmc_size_16_color_FFFFFF_t_70][]

The Horizontal Pod Autoscaler is implemented as a control loop, with a period controlled by the controller manager’s `--horizontal-pod-autoscaler-sync-period` flag (with a default value of 30 seconds).

对于--horizontal-pod-autoscaler-sync-period默认为15s

During each period, the controller manager queries the resource utilization against the metrics specified in each HorizontalPodAutoscaler definition. The controller manager obtains the metrics from either the resource metrics API (for per-pod resource metrics), or the custom metrics API (for all other metrics).

*  For per-pod resource metrics (like CPU), the controller fetches the metrics from the resource metrics API for each pod targeted by the HorizontalPodAutoscaler. Then, if a target utilization value is set, the controller calculates the utilization value as a percentage of the equivalent resource request on the containers in each pod. If a target raw value is set, the raw metric values are used directly. The controller then takes the mean of the utilization or the raw value (depending on the type of target specified) across all targeted pods, and produces a ratio used to scale the number of desired replicas.

The HorizontalPodAutoscaler normally fetches metrics from a series of aggregated APIs (`metrics.k8s.io`, `custom.metrics.k8s.io`, and `external.metrics.k8s.io`). The `metrics.k8s.io` API is usually provided by metrics-server, which needs to be launched separately. See [metrics-server][] for instructions. The HorizontalPodAutoscaler can also fetch metrics directly from Heapster.

###     获取metrics的两种方式: ###

*  Heapster:  直接从heapster获得metrics
 *  aggregated APIs:  包括`metrics.k8s.io`, `custom.metrics.k8s.io`, and `external.metrics.k8s.io`

# 参数： #

配置需要enable的列表，這裏也包括horizontalpodautoscaling

--controllers strings：
    
             All controllers: attachdetach, bootstrapsigner, clusterrole-aggregation, cronjob, csrapproving,
                    csrcleaner, csrsigning, daemonset, deployment, disruption, endpoint, garbagecollector,
                    horizontalpodautoscaling, job, namespace, nodeipam, nodelifecycle, persistentvolume-binder,
                    persistentvolume-expander, podgc, pv-protection, pvc-protection, replicaset, replicationcontroller,
                    resourcequota, route, service, serviceaccount, serviceaccount-token, statefulset, tokencleaner, ttl,
                    ttl-after-finished
                    Disabled-by-default controllers: bootstrapsigner, tokencleaner (default [*])

--horizontal-pod-autoscaler-cpu-initialization-period duration   The period after pod start when CPU samples might be skipped. (default 5m0s)
          --horizontal-pod-autoscaler-downscale-stabilization duration     The period for which autoscaler will look backwards and not scale down below any recommendation it made during that period. (default 5m0s)
          --horizontal-pod-autoscaler-initial-readiness-delay duration     The period after pod start during which readiness changes will be treated as initial readiness. (default 30s)
          --horizontal-pod-autoscaler-sync-period duration                 The period for syncing the number of pods in horizontal pod autoscaler. (default 15s)
          --horizontal-pod-autoscaler-tolerance float                      The minimum change (from 1.0) in the desired-to-actual metrics ratio for the horizontal pod autoscaler to consider scaling. (default 0.1)

# **0. 入口NewControllerInitializers函数** #

注册HPA，controllers\["horizontalpodautoscaling"\] = startHPAController  
 = startTTLAfterFinishedController

// NewControllerInitializers is a public map of named controller groups (you can start more than one in an init func)
    // paired to their InitFunc.  This allows for structured downstream composition and subdivision.
    func NewControllerInitializers(loopMode ControllerLoopMode) map[string]InitFunc {
    	controllers := map[string]InitFunc{}
    	
    	controllers["replicaset"] = startReplicaSetController
    	controllers["horizontalpodautoscaling"] = startHPAController
     = startTTLAfterFinishedController
    
    	return controllers
    }

# 1. 函数startHPAController #

Group为autoscaling，Version为v1，Resource为horizontalpodautoscalers

根据启动参数 --horizontal-pod-autoscaler-use-rest-clients="true"，如果要使用自定义metrics，需要开启该选项，则调用startHPAControllerWithRESTClient函数

否则调用startHPAControllerWithLegacyClient从heapster拉数据

func startHPAController(ctx ControllerContext) (http.Handler, bool, error) {
    	if !ctx.AvailableResources[schema.GroupVersionResource{Group: "autoscaling", Version: "v1", Resource: "horizontalpodautoscalers"}] {
    		return nil, false, nil
    	}
    
    	if ctx.ComponentConfig.HPAController.HorizontalPodAutoscalerUseRESTClients {
    		// use the new-style clients if support for custom metrics is enabled
    		return startHPAControllerWithRESTClient(ctx)
    	}
    
    	return startHPAControllerWithLegacyClient(ctx)
    }

###   1.1 startHPAControllerWithRESTClient函数 ###

調用NewAvailableAPIsGetter建立客戶端連接

PeriodicallyInvalidate函數定期處理不合法的cache

kubernetes 1.12.1 已经启动设置 --horizontal-pod-autoscaler-use-rest-clients="true"

這個函數主要是生成了一大堆client連接

startHPAControllerWithMetricsClient函數做的事情比較多，let me瞅瞅

func startHPAControllerWithRESTClient(ctx ControllerContext) (http.Handler, bool, error) {
    	clientConfig := ctx.ClientBuilder.ConfigOrDie("horizontal-pod-autoscaler")
    	hpaClient := ctx.ClientBuilder.ClientOrDie("horizontal-pod-autoscaler")
    
    	apiVersionsGetter := custom_metrics.NewAvailableAPIsGetter(hpaClient.Discovery())
    	// invalidate the discovery information roughly once per resync interval our API
    	// information is *at most* two resync intervals old.
    	go custom_metrics.PeriodicallyInvalidate(
    		apiVersionsGetter,
    		ctx.ComponentConfig.HPAController.HorizontalPodAutoscalerSyncPeriod.Duration,
    		ctx.Stop)
    
    	metricsClient := metrics.NewRESTMetricsClient(
    		resourceclient.NewForConfigOrDie(clientConfig),
    		custom_metrics.NewForConfig(clientConfig, ctx.RESTMapper, apiVersionsGetter),
    		external_metrics.NewForConfigOrDie(clientConfig),
    	)
    	return startHPAControllerWithMetricsClient(ctx, metricsClient)
    }

**  1.1.1  startHPAControllerWithMetricsClient函數**

NewHorizontalController函數初始化(第2章节讲解)，informer机制包括HPA与pod

func startHPAControllerWithMetricsClient(ctx ControllerContext, metricsClient metrics.MetricsClient) (http.Handler, bool, error) {
    	hpaClient := ctx.ClientBuilder.ClientOrDie("horizontal-pod-autoscaler")
    	hpaClientConfig := ctx.ClientBuilder.ConfigOrDie("horizontal-pod-autoscaler")
    
    	// we don't use cached discovery because DiscoveryScaleKindResolver does its own caching,
    	// so we want to re-fetch every time when we actually ask for it
    	scaleKindResolver := scale.NewDiscoveryScaleKindResolver(hpaClient.Discovery())
    	scaleClient, err := scale.NewForConfig(hpaClientConfig, ctx.RESTMapper, dynamic.LegacyAPIPathResolverFunc, scaleKindResolver)
    	if err != nil {
    		return nil, false, err
    	}
    
    	go podautoscaler.NewHorizontalController(
    		hpaClient.CoreV1(),
    		scaleClient,
    		hpaClient.AutoscalingV1(),
    		ctx.RESTMapper,
    		metricsClient,
    		ctx.InformerFactory.Autoscaling().V1().HorizontalPodAutoscalers(),
    		ctx.InformerFactory.Core().V1().Pods(),
    		ctx.ComponentConfig.HPAController.HorizontalPodAutoscalerSyncPeriod.Duration,
    		ctx.ComponentConfig.HPAController.HorizontalPodAutoscalerDownscaleStabilizationWindow.Duration,
    		ctx.ComponentConfig.HPAController.HorizontalPodAutoscalerTolerance,
    		ctx.ComponentConfig.HPAController.HorizontalPodAutoscalerCPUInitializationPeriod.Duration,
    		ctx.ComponentConfig.HPAController.HorizontalPodAutoscalerInitialReadinessDelay.Duration,
    	).Run(ctx.Stop)
    	return nil, true, nil
    }

###   1.2 startHPAControllerWithMetricsClient函数 ###

###   1.3 startHPAControllerWithMetricsClient函数 ###

# 2. NewHorizontalController函数 #

创建`eventBroadcaster`并设置属性

创建hpa informer机制，回调函数为AddFunc为hpaController.enqueueHPA, UpdateFunc为hpaController.updateHPA，DeleteFunc为hpaController.deleteHPA

NewReplicaCalculator副本计算器

// NewHorizontalController creates a new HorizontalController.
    func NewHorizontalController(
    	evtNamespacer v1core.EventsGetter,
    	scaleNamespacer scaleclient.ScalesGetter,
    	hpaNamespacer autoscalingclient.HorizontalPodAutoscalersGetter,
    	mapper apimeta.RESTMapper,
    	metricsClient metricsclient.MetricsClient,
    	hpaInformer autoscalinginformers.HorizontalPodAutoscalerInformer,
    	podInformer coreinformers.PodInformer,
    	resyncPeriod time.Duration,
    	downscaleStabilisationWindow time.Duration,
    	tolerance float64,
    	cpuInitializationPeriod,
    	delayOfInitialReadinessStatus time.Duration,
    
    ) *HorizontalController {
    	broadcaster := record.NewBroadcaster()
    	broadcaster.StartLogging(glog.Infof)
    	broadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: evtNamespacer.Events("")})
    	recorder := broadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "horizontal-pod-autoscaler"})
    
    	hpaController := &HorizontalController{
    		eventRecorder:                recorder,
    		scaleNamespacer:              scaleNamespacer,
    		hpaNamespacer:                hpaNamespacer,
    		downscaleStabilisationWindow: downscaleStabilisationWindow,
    		queue:           workqueue.NewNamedRateLimitingQueue(NewDefaultHPARateLimiter(resyncPeriod), "horizontalpodautoscaler"),
    		mapper:          mapper,
    		recommendations: map[string][]timestampedRecommendation{},
    	}
    
    	hpaInformer.Informer().AddEventHandlerWithResyncPeriod(
    		cache.ResourceEventHandlerFuncs{
    			AddFunc:    hpaController.enqueueHPA,
    			UpdateFunc: hpaController.updateHPA,
    			DeleteFunc: hpaController.deleteHPA,
    		},
    		resyncPeriod,
    	)
    	hpaController.hpaLister = hpaInformer.Lister()
    	hpaController.hpaListerSynced = hpaInformer.Informer().HasSynced
    
    	hpaController.podLister = podInformer.Lister()
    	hpaController.podListerSynced = podInformer.Informer().HasSynced
    
    	replicaCalc := NewReplicaCalculator(
    		metricsClient,
    		hpaController.podLister,
    		tolerance,
    		cpuInitializationPeriod,
    		delayOfInitialReadinessStatus,
    	)
    	hpaController.replicaCalc = replicaCalc
    
    	return hpaController
    }

# 3. Run函数 #

HPA进行watch以及进行sync

// Run begins watching and syncing.
    func (a *HorizontalController) Run(stopCh <-chan struct{}) {
    	defer utilruntime.HandleCrash()
    	defer a.queue.ShutDown()
    
    	glog.Infof("Starting HPA controller")
    	defer glog.Infof("Shutting down HPA controller")
    
    	if !controller.WaitForCacheSync("HPA", stopCh, a.hpaListerSynced, a.podListerSynced) {
    		return
    	}
    
    	// start a single worker (we may wish to start more in the future)
    	go wait.Until(a.worker, time.Second, stopCh)
    
    	<-stopCh
    }

###   3.1 worker工作了 ###

从queue取key，并进行处理

一直跟踪跟踪，reconcileAutoscaler函数进行自动调整（第4章节讲解）

func (a *HorizontalController) worker() {
    	for a.processNextWorkItem() {
    	}
    	glog.Infof("horizontal pod autoscaler controller worker shutting down")
    }
    
    func (a *HorizontalController) processNextWorkItem() bool {
    	key, quit := a.queue.Get()
    	if quit {
    		return false
    	}
    	defer a.queue.Done(key)
    
    	err := a.reconcileKey(key.(string))
    	if err == nil {
    		// don't "forget" here because we want to only process a given HPA once per resync interval
    		return true
    	}
    
    	a.queue.AddRateLimited(key)
    	utilruntime.HandleError(err)
    	return true
    }

# 4. reconcileAutoscaler函数 #

通过算法调节当前副本数，期望副本数，边界(最大最小)副本数三者的关系

###   4.1 这一沱主要是处理版本问题 ###

// make a copy so that we never mutate the shared informer cache (conversion can mutate the object)
    	hpav1 := hpav1Shared.DeepCopy()
    	// then, convert to autoscaling/v2, which makes our lives easier when calculating metrics
    	hpaRaw, err := unsafeConvertToVersionVia(hpav1, autoscalingv2.SchemeGroupVersion)
    	if err != nil {
    		a.eventRecorder.Event(hpav1, v1.EventTypeWarning, "FailedConvertHPA", err.Error())
    		return fmt.Errorf("failed to convert the given HPA to %s: %v", autoscalingv2.SchemeGroupVersion.String(), err)
    	}
    	hpa := hpaRaw.(*autoscalingv2.HorizontalPodAutoscaler)
    	hpaStatusOriginal := hpa.Status.DeepCopy()
    
    	reference := fmt.Sprintf("%s/%s/%s", hpa.Spec.ScaleTargetRef.Kind, hpa.Namespace, hpa.Spec.ScaleTargetRef.Name)
    
    	targetGV, err := schema.ParseGroupVersion(hpa.Spec.ScaleTargetRef.APIVersion)
    	if err != nil {
    		a.eventRecorder.Event(hpa, v1.EventTypeWarning, "FailedGetScale", err.Error())
    		setCondition(hpa, autoscalingv2.AbleToScale, v1.ConditionFalse, "FailedGetScale", "the HPA controller was unable to get the target's current scale: %v", err)
    		a.updateStatusIfNeeded(hpaStatusOriginal, hpa)
    		return fmt.Errorf("invalid API version in scale target reference: %v", err)
    	}

###   4.2 replicas为0的情况 ###

设置desiredReplicas为0，不进行scale，事件ScalingDisabled

设置最小最大副本数上下限，当前副本最小为1

if scale.Spec.Replicas == 0 {
    		// Autoscaling is disabled for this resource
    		desiredReplicas = 0
    		rescale = false
    		setCondition(hpa, autoscalingv2.ScalingActive, v1.ConditionFalse, "ScalingDisabled", "scaling is disabled since the replica count of the target is zero")
    	} else if currentReplicas > hpa.Spec.MaxReplicas {
    		rescaleReason = "Current number of replicas above Spec.MaxReplicas"
    		desiredReplicas = hpa.Spec.MaxReplicas
    	} else if hpa.Spec.MinReplicas != nil && currentReplicas < *hpa.Spec.MinReplicas {
    		rescaleReason = "Current number of replicas below Spec.MinReplicas"
    		desiredReplicas = *hpa.Spec.MinReplicas
    	} else if currentReplicas == 0 {
    		rescaleReason = "Current number of replicas must be greater than 0"
    		desiredReplicas = 1
    	}

###   4.3 computeReplicasForMetrics函数 ###

根据metrics指标计算对应的副本数 （第5章节讲解）

} else {
    
    		metricDesiredReplicas, metricName, metricStatuses, metricTimestamp, err = a.computeReplicasForMetrics(hpa, scale, hpa.Spec.Metrics)
    		if err != nil {
    			a.setCurrentReplicasInStatus(hpa, currentReplicas)
    			if err := a.updateStatusIfNeeded(hpaStatusOriginal, hpa); err != nil {
    				utilruntime.HandleError(err)
    			}
    			a.eventRecorder.Event(hpa, v1.EventTypeWarning, "FailedComputeMetricsReplicas", err.Error())
    			return fmt.Errorf("failed to compute desired number of replicas based on listed metrics for %s: %v", reference, err)
    		}

###   4.4 比较进行处理 ###

rescaleMetric := ""
    		if metricDesiredReplicas > desiredReplicas {
    			desiredReplicas = metricDesiredReplicas
    			timestamp = metricTimestamp
    			rescaleMetric = metricName
    		}
    		if desiredReplicas > currentReplicas {
    			rescaleReason = fmt.Sprintf("%s above target", rescaleMetric)
    		}
    		if desiredReplicas < currentReplicas {
    			rescaleReason = "All metrics below target"
    		}
    		desiredReplicas = a.normalizeDesiredReplicas(hpa, key, currentReplicas, desiredReplicas)
    		rescale = desiredReplicas != currentReplicas

###   4.5 期待副本数和当前副本数不等，需要扩容或者缩容 ###

調用scales.Update函數，实现在vendor/k8s.io/client-go/scale/client.go中，这个最终会修改deploy或者rc副本数，然后交给deployment或者rc controller处理，deploy实际是由rs处理副本数量，第8章节讲解

if rescale {
    		scale.Spec.Replicas = desiredReplicas
    		_, err = a.scaleNamespacer.Scales(hpa.Namespace).Update(targetGR, scale)
    		if err != nil {
    			a.eventRecorder.Eventf(hpa, v1.EventTypeWarning, "FailedRescale", "New size: %d; reason: %s; error: %v", desiredReplicas, rescaleReason, err.Error())
    			setCondition(hpa, autoscalingv2.AbleToScale, v1.ConditionFalse, "FailedUpdateScale", "the HPA controller was unable to update the target scale: %v", err)
    			a.setCurrentReplicasInStatus(hpa, currentReplicas)
    			if err := a.updateStatusIfNeeded(hpaStatusOriginal, hpa); err != nil {
    				utilruntime.HandleError(err)
    			}
    			return fmt.Errorf("failed to rescale %s: %v", reference, err)
    		}
    		setCondition(hpa, autoscalingv2.AbleToScale, v1.ConditionTrue, "SucceededRescale", "the HPA controller was able to update the target scale to %d", desiredReplicas)
    		a.eventRecorder.Eventf(hpa, v1.EventTypeNormal, "SuccessfulRescale", "New size: %d; reason: %s", desiredReplicas, rescaleReason)
    		glog.Infof("Successful rescale of %s, old size: %d, new size: %d, reason: %s",
    			hpa.Name, currentReplicas, desiredReplicas, rescaleReason)
    	} else {
    		glog.V(4).Infof("decided not to scale %s to %v (last scale time was %s)", reference, desiredReplicas, hpa.Status.LastScaleTime)
    		desiredReplicas = currentReplicas
    	}

# 5. computeReplicasForMetrics函数 #

###   5.1 selector主要是deploy中的labels，主要是过滤功能 ###

for i, metricSpec := range metricSpecs {
    		if scale.Status.Selector == "" {
    			errMsg := "selector is required"
    			a.eventRecorder.Event(hpa, v1.EventTypeWarning, "SelectorRequired", errMsg)
    			setCondition(hpa, autoscalingv2.ScalingActive, v1.ConditionFalse, "InvalidSelector", "the HPA target's scale is missing a selector")
    			return 0, "", nil, time.Time{}, fmt.Errorf(errMsg)
    		}
    
    		selector, err := labels.Parse(scale.Status.Selector)
    selector)
    		if err != nil {
    			errMsg := fmt.Sprintf("couldn't convert selector into a corresponding internal selector object: %v", err)
    			a.eventRecorder.Event(hpa, v1.EventTypeWarning, "InvalidSelector", errMsg)
    			setCondition(hpa, autoscalingv2.ScalingActive, v1.ConditionFalse, "InvalidSelector", errMsg)
    			return 0, "", nil, time.Time{}, fmt.Errorf(errMsg)
    		}

###   5.2 metric类型为Resource类型 ###

挑一个resource类型讲解，调用computeStatusForResourceMetric，计算资源

switch metricSpec.Type {
    		case autoscalingv2.ObjectMetricSourceType:
    			metricSelector, err := metav1.LabelSelectorAsSelector(metricSpec.Object.Metric.Selector)
    			if err != nil {
    				a.eventRecorder.Event(hpa, v1.EventTypeWarning, "FailedGetObjectMetric", err.Error())
    				setCondition(hpa, autoscalingv2.ScalingActive, v1.ConditionFalse, "FailedGetObjectMetric", "the HPA was unable to compute the replica count: %v", err)
    				return 0, "", nil, time.Time{}, fmt.Errorf("failed to get object metric value: %v", err)
    			}
    			replicaCountProposal, timestampProposal, metricNameProposal, err = a.computeStatusForObjectMetric(currentReplicas, metricSpec, hpa, selector, &statuses[i], metricSelector)
    			if err != nil {
    				return 0, "", nil, time.Time{}, fmt.Errorf("failed to get object metric value: %v", err)
    			}
    
            case autoscalingv2.ResourceMetricSourceType:
    			replicaCountProposal, timestampProposal, metricNameProposal, err = a.computeStatusForResourceMetric(currentReplicas, metricSpec, hpa, selector, &statuses[i])
    			if err != nil {
    				return 0, "", nil, time.Time{}, err
    			}

###   5.3  如果预估的大于当前副本，则更新 ###

if replicas == 0 || replicaCountProposal > replicas {
    			timestamp = timestampProposal
    			replicas = replicaCountProposal
    			metric = metricNameProposal
    		}

** **

# **6. **computeStatusForResourceMetric**函数** #

**      计算想要期待的副本数，根据是否设置**AverageValue处理，本文基于未设置讲解，其流程基本类似

###   6.1 基于未设置AverageValue流程 ###

调用GetResourceReplicas计算期待运行的副本数

} else {
    		if metricSpec.Resource.Target.AverageUtilization == nil {
    			errMsg := "invalid resource metric source: neither a utilization target nor a value target was set"
    			a.eventRecorder.Event(hpa, v1.EventTypeWarning, "FailedGetResourceMetric", errMsg)
    			setCondition(hpa, autoscalingv2.ScalingActive, v1.ConditionFalse, "FailedGetResourceMetric", "the HPA was unable to compute the replica count: %s", errMsg)
    			return 0, time.Time{}, "", fmt.Errorf(errMsg)
    		}
    		targetUtilization := *metricSpec.Resource.Target.AverageUtilization
    		var percentageProposal int32
    		var rawProposal int64
    
    		replicaCountProposal, percentageProposal, rawProposal, timestampProposal, err := a.replicaCalc.GetResourceReplicas(currentReplicas, targetUtilization, metricSpec.Resource.Name, hpa.Namespace, selector)
    		if err != nil {
    			a.eventRecorder.Event(hpa, v1.EventTypeWarning, "FailedGetResourceMetric", err.Error())
    			setCondition(hpa, autoscalingv2.ScalingActive, v1.ConditionFalse, "FailedGetResourceMetric", "the HPA was unable to compute the replica count: %v", err)
    			return 0, time.Time{}, "", fmt.Errorf("failed to get %s utilization: %v", metricSpec.Resource.Name, err)
    		}
    		metricNameProposal := fmt.Sprintf("%s resource utilization (percentage of request)", metricSpec.Resource.Name)
    		*status = autoscalingv2.MetricStatus{
    			Type: autoscalingv2.ResourceMetricSourceType,
    			Resource: &autoscalingv2.ResourceMetricStatus{
    				Name: metricSpec.Resource.Name,
    				Current: autoscalingv2.MetricValueStatus{
    					AverageUtilization: &percentageProposal,
    					AverageValue:       resource.NewMilliQuantity(rawProposal, resource.DecimalSI),
    				},
    			},
    		}
    		return replicaCountProposal, timestampProposal, metricNameProposal, nil
    	}

###   6.2 填充status结构体， ###

metricNameProposal := fmt.Sprintf("%s resource utilization (percentage of request)", metricSpec.Resource.Name)
    		*status = autoscalingv2.MetricStatus{
    			Type: autoscalingv2.ResourceMetricSourceType,
    			Resource: &autoscalingv2.ResourceMetricStatus{
    				Name: metricSpec.Resource.Name,
    				Current: autoscalingv2.MetricValueStatus{
    					AverageUtilization: &percentageProposal,
    					AverageValue:       resource.NewMilliQuantity(rawProposal, resource.DecimalSI),
    				},
    			},
    		}

# 7. GetResourceReplicas函数 #

###   7.1 GetResourceMetric函数 ###

metrics, timestamp, err := c.metricsClient.GetResourceMetric(resource, namespace, selector)
    	if err != nil {
    		return 0, 0, 0, time.Time{}, fmt.Errorf("unable to get metrics for resource %s: %v", resource, err)
    	}

**  7.1.1 GetResourceMetric函数**

主要是根据标签过滤调用List API，对各项metrics指标将所有容器指标项加和计算

// GetResourceMetric gets the given resource metric (and an associated oldest timestamp)
    // for all pods matching the specified selector in the given namespace
    func (c *resourceMetricsClient) GetResourceMetric(resource v1.ResourceName, namespace string, selector labels.Selector) (PodMetricsInfo, time.Time, error) {
    	metrics, err := c.client.PodMetricses(namespace).List(metav1.ListOptions{LabelSelector: selector.String()})
    	if err != nil {
    		return nil, time.Time{}, fmt.Errorf("unable to fetch metrics from resource metrics API: %v", err)
    	}
    
    	if len(metrics.Items) == 0 {
    		return nil, time.Time{}, fmt.Errorf("no metrics returned from resource metrics API")
    	}
    
    	res := make(PodMetricsInfo, len(metrics.Items))
    
    	for _, m := range metrics.Items {
    		podSum := int64(0)
    		missing := len(m.Containers) == 0
    		for _, c := range m.Containers {
    			resValue, found := c.Usage[v1.ResourceName(resource)]
    			if !found {
    				missing = true
    				glog.V(2).Infof("missing resource metric %v for container %s in pod %s/%s", resource, c.Name, namespace, m.Name)
    				break // containers loop
    			}
    			podSum += resValue.MilliValue()
    		}
    
    		if !missing {
    			res[m.Name] = PodMetric{
    				Timestamp: m.Timestamp.Time,
    				Window:    m.Window.Duration,
    				Value:     int64(podSum),
    			}
    		}
    	}
    
    	timestamp := metrics.Items[0].Timestamp.Time
    
    	return res, timestamp, nil
    }

###   7.2 调用API获得pod，并对pod分组 ###

*   ignoredPods: condition == nil || pod.Status.StartTime == nil
 *  missingPods:

把忽略的pod从metrics的pod中删除

podList, err := c.podLister.Pods(namespace).List(selector)
    	if err != nil {
    		return 0, 0, 0, time.Time{}, fmt.Errorf("unable to get pods while calculating replica count: %v", err)
    	}
    
    	itemsLen := len(podList)
    	if itemsLen == 0 {
    		return 0, 0, 0, time.Time{}, fmt.Errorf("no pods returned by selector while calculating replica count")
    	}
    
    	readyPodCount, ignoredPods, missingPods := groupPods(podList, metrics, resource, c.cpuInitializationPeriod, c.delayOfInitialReadinessStatus)
    	removeMetricsForPods(metrics, ignoredPods)

###   7.3 calculatePodRequests函数 ###

计算所有pod名下的容器指标累加和

func calculatePodRequests(pods []*v1.Pod, resource v1.ResourceName) (map[string]int64, error) {
    	requests := make(map[string]int64, len(pods))
    	for _, pod := range pods {
    		podSum := int64(0)
    		for _, container := range pod.Spec.Containers {
    			if containerRequest, ok := container.Resources.Requests[resource]; ok {
    				podSum += containerRequest.MilliValue()
    			} else {
    				return nil, fmt.Errorf("missing request for %s", resource)
    			}
    		}
    		requests[pod.Name] = podSum
    	}
    	return requests, nil
    }

###   7.4 GeteResourceUtilizationRatio函数 ###

计算资源指标利用率

当前利用率 = int32((metricsTotal \* 100) / requestsTotal)，所有资源和 / 所有pod request和

// GetResourceUtilizationRatio takes in a set of metrics, a set of matching requests,
    // and a target utilization percentage, and calculates the ratio of
    // desired to actual utilization (returning that, the actual utilization, and the raw average value)
    func GetResourceUtilizationRatio(metrics PodMetricsInfo, requests map[string]int64, targetUtilization int32) (utilizationRatio float64, currentUtilization int32, rawAverageValue int64, err error) {
    	metricsTotal := int64(0)
    	requestsTotal := int64(0)
    	numEntries := 0
    
    	for podName, metric := range metrics {
    		request, hasRequest := requests[podName]
    		if !hasRequest {
    			// we check for missing requests elsewhere, so assuming missing requests == extraneous metrics
    			continue
    		}
    
    		metricsTotal += metric.Value
    		requestsTotal += request
    		numEntries++
    	}
    
    	// if the set of requests is completely disjoint from the set of metrics,
    	// then we could have an issue where the requests total is zero
    	if requestsTotal == 0 {
    		return 0, 0, 0, fmt.Errorf("no metrics returned matched known pods")
    	}
    
    	currentUtilization = int32((metricsTotal * 100) / requestsTotal)
    
    	return float64(currentUtilization) / float64(targetUtilization), currentUtilization, metricsTotal / int64(numEntries), nil
    }

###   7.5 计算需要使用的副本数 ###

\--horizontal-pod-autoscaler-tolerance float： The minimum change (from 1.0) in the desired-to-actual metrics ratio for the horizontal pod autoscaler to consider scaling. (default 0.1)

如果使用率可以在容忍的范围内，就是还是使用当前副本数，否则使用使用率 \* readyPodCount作为新副本数

rebalanceIgnored := len(ignoredPods) > 0 && usageRatio > 1.0
    	if !rebalanceIgnored && len(missingPods) == 0 {
    		if math.Abs(1.0-usageRatio) <= c.tolerance {
    			// return the current replicas if the change would be too small
    			return currentReplicas, utilization, rawUtilization, 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, rawUtilization, 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] = metricsclient.PodMetric{Value: requests[podName]}
    			}
    		} else if usageRatio > 1.0 {
    			// on a scale-up, treat missing pods as using 0% of the resource request
    			for podName := range missingPods {
    				metrics[podName] = metricsclient.PodMetric{Value: 0}
    			}
    		}
    	}
    
    	if rebalanceIgnored {
    		// on a scale-up, treat unready pods as using 0% of the resource request
    		for podName := range ignoredPods {
    			metrics[podName] = metricsclient.PodMetric{Value: 0}
    		}
    	}
    
    	// re-run the utilization calculation with our new numbers
    	newUsageRatio, _, _, err := metricsclient.GetResourceUtilizationRatio(metrics, requests, targetUtilization)
    	if err != nil {
    		return 0, utilization, rawUtilization, time.Time{}, err
    	}
    
    	if math.Abs(1.0-newUsageRatio) <= c.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, rawUtilization, 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, rawUtilization, timestamp, nil

# 8. 初始化scaleClient #

startHPAControllerWithMetricsClient函数中调用NewForConfig初始化

// we don't use cached discovery because DiscoveryScaleKindResolver does its own caching,
    	// so we want to re-fetch every time when we actually ask for it
    	scaleKindResolver := scale.NewDiscoveryScaleKindResolver(hpaClient.Discovery())
    	scaleClient, err := scale.NewForConfig(hpaClientConfig, ctx.RESTMapper, dynamic.LegacyAPIPathResolverFunc, scaleKindResolver)
    	if err != nil {
    		return nil, false, err
    	}

###   8.1 NewForConfig函数 ###

路径 vendor/k8s.io/client-go/scale/client.go

// NewForConfig creates a new ScalesGetter which resolves kinds
    // to resources using the given RESTMapper, and API paths using
    // the given dynamic.APIPathResolverFunc.
    func NewForConfig(cfg *restclient.Config, mapper PreferredResourceMapper, resolver dynamic.APIPathResolverFunc, scaleKindResolver ScaleKindResolver) (ScalesGetter, error) {
    	// so that the RESTClientFor doesn't complain
    	cfg.GroupVersion = &schema.GroupVersion{}
    
    	cfg.NegotiatedSerializer = serializer.DirectCodecFactory{
    		CodecFactory: codecs,
    	}
    	if len(cfg.UserAgent) == 0 {
    		cfg.UserAgent = restclient.DefaultKubernetesUserAgent()
    	}
    
    	client, err := restclient.RESTClientFor(cfg)
    	if err != nil {
    		return nil, err
    	}
    
    	return New(client, mapper, resolver, scaleKindResolver), nil
    }

###  8.2 scaleClient结构体 ###

实现了ScaleInterface接口

// scaleClient is an implementation of ScalesGetter
    // which makes use of a RESTMapper and a generic REST
    // client to support an discoverable resource.
    // It behaves somewhat similarly to the dynamic ClientPool,
    // but is more specifically scoped to Scale.
    type scaleClient struct {
    	mapper PreferredResourceMapper
    
    	apiPathResolverFunc dynamic.APIPathResolverFunc
    	scaleKindResolver   ScaleKindResolver
    	clientBase          restclient.Interface
    }
    
    // ScalesGetter can produce a ScaleInterface
    // for a particular namespace.
    type ScalesGetter interface {
    	Scales(namespace string) ScaleInterface
    }
    
    // ScaleInterface can fetch and update scales for
    // resources in a particular namespace which implement
    // the scale subresource.
    type ScaleInterface interface {
    	// Get fetches the scale of the given scalable resource.
    	Get(resource schema.GroupResource, name string) (*autoscalingapi.Scale, error)
    
    	// Update updates the scale of the the given scalable resource.
    	Update(resource schema.GroupResource, scale *autoscalingapi.Scale) (*autoscalingapi.Scale, error)
    }

###   8.3 Update函数 ###

最终还是调用的API

func (c *namespacedScaleClient) Update(resource schema.GroupResource, scale *autoscaling.Scale) (*autoscaling.Scale, error) {
    	...............
    
    	result := c.client.clientBase.Put().
    		AbsPath(path).
    		Namespace(c.namespace).
    		Resource(gvr.Resource).
    		Name(scale.Name).
    		SubResource("scale").
    		Body(scaleUpdateBytes).
    		Do()
    	................
    	return scaleObj.(*autoscaling.Scale), err
    }

**  8.3.1 对于deployment注册**

func (r *ScaleREST) GroupVersionKind(containingGV schema.GroupVersion) schema.GroupVersionKind {
    	switch containingGV {
    	case extensionsv1beta1.SchemeGroupVersion:
    		return extensionsv1beta1.SchemeGroupVersion.WithKind("Scale")
    	case appsv1beta1.SchemeGroupVersion:
    		return appsv1beta1.SchemeGroupVersion.WithKind("Scale")
    	case appsv1beta2.SchemeGroupVersion:
    		return appsv1beta2.SchemeGroupVersion.WithKind("Scale")
    	default:
    		return autoscalingv1.SchemeGroupVersion.WithKind("Scale")
    	}
    }

###  8.4 Update函数 ###

路径 pkg/registry/apps/deployment/storage/storage.go

修改了replicas，等待controller取处理

func (r *ScaleREST) Update(ctx context.Context, name string, objInfo rest.UpdatedObjectInfo, createValidation rest.ValidateObjectFunc, updateValidation rest.ValidateObjectUpdateFunc, forceAllowCreate bool, options *metav1.UpdateOptions) (runtime.Object, bool, error) {
    	obj, err := r.store.Get(ctx, name, &metav1.GetOptions{})
    	if err != nil {
    		return nil, false, errors.NewNotFound(extensions.Resource("deployments/scale"), name)
    	}
    	deployment := obj.(*extensions.Deployment)
    
    	oldScale, err := scaleFromDeployment(deployment)
    	if err != nil {
    		return nil, false, err
    	}
    
    	obj, err = objInfo.UpdatedObject(ctx, oldScale)
    	if err != nil {
    		return nil, false, err
    	}
    	if obj == nil {
    		return nil, false, errors.NewBadRequest(fmt.Sprintf("nil update passed to Scale"))
    	}
    	scale, ok := obj.(*autoscaling.Scale)
    	if !ok {
    		return nil, false, errors.NewBadRequest(fmt.Sprintf("expected input object type to be Scale, but %T", obj))
    	}
    
    	if errs := autoscalingvalidation.ValidateScale(scale); len(errs) > 0 {
    		return nil, false, errors.NewInvalid(extensions.Kind("Scale"), name, errs)
    	}
    
    	deployment.Spec.Replicas = scale.Spec.Replicas
    	deployment.ResourceVersion = scale.ResourceVersion
    	obj, _, err = r.store.Update(ctx, deployment.Name, rest.DefaultUpdatedObjectInfo(deployment), createValidation, updateValidation, false, options)
    	if err != nil {
    		return nil, false, err
    	}
    	deployment = obj.(*extensions.Deployment)
    	newScale, err := scaleFromDeployment(deployment)
    	if err != nil {
    		return nil, false, errors.NewBadRequest(fmt.Sprintf("%v", err))
    	}
    	return newScale, false, nil
    }

[https_kubernetes.io_docs_tasks_run-application_horizontal-pod-autoscale]: https://kubernetes.io/docs/tasks/run-application/horizontal-pod-autoscale/
[custom metrics]: https://git.k8s.io/community/contributors/design-proposals/instrumentation/custom-metrics-api.md
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob25nbGluemhhbmc_size_16_color_FFFFFF_t_70]: /images/20220412/dd8bd51ebe6b43699132b1e654e681da.png
[metrics-server]: https://kubernetes.io/docs/tasks/debug-application-cluster/core-metrics-pipeline/#metrics-server