K8s hpa 控制器中的代码实现（二）

hpaController := &HorizontalController{

    queue: workqueue.NewTypedRateLimitingQueueWithConfig(
        NewDefaultHPARateLimiter(resyncPeriod),
        workqueue.TypedRateLimitingQueueConfig[string]{
            Name: "horizontalpodautoscaler",
        },
    ),

}

func NewFixedItemIntervalRateLimiter(interval time.Duration) workqueue.TypedRateLimiter[string] {
	return &FixedItemIntervalRateLimiter{
		interval: interval,
	}
}

if resource == v1.ResourceCPU {
    var unready bool
    _, condition := podutil.GetPodCondition(&pod.Status, v1.PodReady)
    if condition == nil || pod.Status.StartTime == nil {
        unready = true
    } else {
        // Pod still within possible initialisation period.
        if pod.Status.StartTime.Add(cpuInitializationPeriod).After(time.Now()) {
            // Ignore sample if pod is unready or one window of metric wasn't collected since last state transition.
            unready = condition.Status == v1.ConditionFalse || metric.Timestamp.Before(condition.LastTransitionTime.Time.Add(metric.Window))
        } else {
            // Ignore metric if pod is unready and it has never been ready.
            unready = condition.Status == v1.ConditionFalse && pod.Status.StartTime.Add(delayOfInitialReadinessStatus).After(condition.LastTransitionTime.Time)
        }
    }
    if unready {
        unreadyPods.Insert(pod.Name)
        continue
    }
}

func (c *ReplicaCalculator) calcPlainMetricReplicas(
	metrics metricsclient.PodMetricsInfo, // pod 的指标，key 是 pod 的 name，value 是 pod 的指标
	currentReplicas int32, // 当前的副本数
	targetUsage int64, // 目标利用率
	tolerances Tolerances, // 容忍度，默认为 nil
	namespace string, // 命名空间
	selector labels.Selector, // pod selector
	resource v1.ResourceName, // 资源类型，假设为 cpu
) (replicaCount int32, usage int64, err error) {

	podList, _ := c.podLister.Pods(namespace).List(selector)

	// 根据上面的 podList，在 metrics 中获取 pod 的指标，并分类：
	// 未就绪的 pod，缺失 metrics的 pod，忽略的 pod
	readyPodCount, unreadyPods, missingPods, ignoredPods := groupPods(podList, metrics, resource, c.cpuInitializationPeriod, c.delayOfInitialReadinessStatus)
	removeMetricsForPods(metrics, ignoredPods)
	removeMetricsForPods(metrics, unreadyPods)

	// 计算指标的利用率: float64(currentUsage) / float64(targetUsage)
	// 如果 usageRatio > 1.0，则表示要扩容
	// 如果 usageRatio < 1.0，则表示要缩容
	usageRatio, usage := metricsclient.GetMetricUsageRatio(metrics, targetUsage)

	// 如果存在未就绪的 pod，并且计算出来的比例是需要扩容
	scaleUpWithUnready := len(unreadyPods) > 0 && usageRatio > 1.0
	if !scaleUpWithUnready && len(missingPods) == 0 {
		if tolerances.isWithin(usageRatio) {
			// 容忍度在 [1-tolerances.scaleDown, 1+tolerances.scaleUp] 之间，则认为当前的副本数是合适的
			return currentReplicas, usage, nil
		}

		// 如果没有任何未就绪的 pod 或者缺失 metrics 的 pod，则可以计算新的副本数
		// NOTE：这个是 NORMAL case
		return int32(math.Ceil(usageRatio * float64(readyPodCount))), usage, nil
	}

	if len(missingPods) > 0 {
		if usageRatio < 1.0 {
			// 在缩容时，认为 metrics 丢失的 pod 的资源使用情况，跟设置的 targetUsage
			// 在这种情况下，metrics 丢失的 pod 不会影响缩容比例
			for podName := range missingPods {
				metrics[podName] = metricsclient.PodMetric{Value: targetUsage}
			}
		} else if usageRatio > 1.0 {
			// 在扩容时，认为 metrics 丢失的 pod 所使用的资源量为 0，这样计算出来的 usageRatio 会偏小
			// 因此扩容会相对保守
			for podName := range missingPods {
				metrics[podName] = metricsclient.PodMetric{Value: 0}
			}
		}
	}

	if scaleUpWithUnready {
		// 在扩容时，认为未就绪的 pod 所使用的资源量为 0，这样计算出来的 usageRatio 会偏小
		// 因此扩容会相对保守
		for podName := range unreadyPods {
			metrics[podName] = metricsclient.PodMetric{Value: 0}
		}
	}

	// 根据上面的调整(missing pod 以及 unready pod)，重新计算 usageRatio
	newUsageRatio, _ := metricsclient.GetMetricUsageRatio(metrics, targetUsage)

	if tolerances.isWithin(newUsageRatio) || (usageRatio < 1.0 && newUsageRatio > 1.0) || (usageRatio > 1.0 && newUsageRatio < 1.0) {
		// 若符合一定的容忍度，则不进行扩缩容
		return currentReplicas, usage, nil
	}

	// 重新计算副本数
	newReplicas := int32(math.Ceil(newUsageRatio * float64(len(metrics))))
	if (newUsageRatio < 1.0 && newReplicas > currentReplicas) || (newUsageRatio > 1.0 && newReplicas < currentReplicas) {
		// 若符合一定的容忍度，则不进行扩缩容
		return currentReplicas, usage, nil
	}

	// 返回结果，副本数是根据 metrics 计算出来的副本数
	return newReplicas, usage, nil
}

// stabilizeRecommendationWithBehaviors:
// - replaces old recommendation with the newest recommendation,
// - returns {max,min} of recommendations that are not older than constraints.Scale{Up,Down}.DelaySeconds
func (a *HorizontalController) stabilizeRecommendationWithBehaviors(args NormalizationArg) (int32, string, string) {
	now := time.Now()

	foundOldSample := false
	oldSampleIndex := 0

	upRecommendation := args.DesiredReplicas
	upDelaySeconds := *args.ScaleUpBehavior.StabilizationWindowSeconds
	upCutoff := now.Add(-time.Second * time.Duration(upDelaySeconds))

	downRecommendation := args.DesiredReplicas
	downDelaySeconds := *args.ScaleDownBehavior.StabilizationWindowSeconds
	downCutoff := now.Add(-time.Second * time.Duration(downDelaySeconds))

	// Calculate the upper and lower stabilization limits.
	a.recommendationsLock.Lock()
	defer a.recommendationsLock.Unlock()
	for i, rec := range a.recommendations[args.Key] {
		if rec.timestamp.After(upCutoff) {
			upRecommendation = min(rec.recommendation, upRecommendation)
		}
		if rec.timestamp.After(downCutoff) {
			downRecommendation = max(rec.recommendation, downRecommendation)
		}
		if rec.timestamp.Before(upCutoff) && rec.timestamp.Before(downCutoff) {
			foundOldSample = true
			oldSampleIndex = i
		}
	}

	// Bring the recommendation to within the upper and lower limits (stabilize).
	recommendation := args.CurrentReplicas
	if recommendation < upRecommendation {
		recommendation = upRecommendation
	}
	if recommendation > downRecommendation {
		recommendation = downRecommendation
	}

	// Record the unstabilized recommendation.
	if foundOldSample {
		a.recommendations[args.Key][oldSampleIndex] = timestampedRecommendation{args.DesiredReplicas, time.Now()}
	} else {
		a.recommendations[args.Key] = append(a.recommendations[args.Key], timestampedRecommendation{args.DesiredReplicas, time.Now()})
	}

	// Determine a human-friendly message.
	var reason, message string
	if args.DesiredReplicas >= args.CurrentReplicas {
		reason = "ScaleUpStabilized"
		message = "recent recommendations were lower than current one, applying the lowest recent recommendation"
	} else {
		reason = "ScaleDownStabilized"
		message = "recent recommendations were higher than current one, applying the highest recent recommendation"
	}
	return recommendation, reason, message
}