cluster-proportional-autoscaler源码分析及如何解决KubeDNS性能瓶颈

pkg/autoscaler/autoscaler_server.go:82
func (s *AutoScaler) pollAPIServer() {
	// Query the apiserver for the cluster status --- number of nodes and cores
	clusterStatus, err := s.k8sClient.GetClusterStatus()
	if err != nil {
		glog.Errorf("Error while getting cluster status: %v", err)
		return
	}
	glog.V(4).Infof("Total nodes %5d, schedulable nodes: %5d", clusterStatus.TotalNodes, clusterStatus.SchedulableNodes)
	glog.V(4).Infof("Total cores %5d, schedulable cores: %5d", clusterStatus.TotalCores, clusterStatus.SchedulableCores)

	// Sync autoscaler ConfigMap with apiserver
	configMap, err := s.syncConfigWithServer()
	if err != nil || configMap == nil {
		glog.Errorf("Error syncing configMap with apiserver: %v", err)
		return
	}

	// Only sync updated ConfigMap or before controller is set.
	if s.controller == nil || configMap.ObjectMeta.ResourceVersion != s.controller.GetParamsVersion() {
		// Ensure corresponding controller type and scaling params.
		s.controller, err = plugin.EnsureController(s.controller, configMap)
		if err != nil || s.controller == nil {
			glog.Errorf("Error ensuring controller: %v", err)
			return
		}
	}

	// Query the controller for the expected replicas number
	expReplicas, err := s.controller.GetExpectedReplicas(clusterStatus)
	if err != nil {
		glog.Errorf("Error calculating expected replicas number: %v", err)
		return
	}
	glog.V(4).Infof("Expected replica count: %3d", expReplicas)

	// Update resource target with expected replicas.
	_, err = s.k8sClient.UpdateReplicas(expReplicas)
	if err != nil {
		glog.Errorf("Update failure: %s", err)
	}
}

pkg/autoscaler/k8sclient/k8sclient.go:142
func (k *k8sClient) GetClusterStatus() (clusterStatus *ClusterStatus, err error) {
	opt := metav1.ListOptions{Watch: false}

	nodes, err := k.clientset.CoreV1().Nodes().List(opt)
	if err != nil || nodes == nil {
		return nil, err
	}
	clusterStatus = &ClusterStatus{}
	clusterStatus.TotalNodes = int32(len(nodes.Items))
	var tc resource.Quantity
	var sc resource.Quantity
	for _, node := range nodes.Items {
		tc.Add(node.Status.Capacity[apiv1.ResourceCPU])
		if !node.Spec.Unschedulable {
			clusterStatus.SchedulableNodes++
			sc.Add(node.Status.Capacity[apiv1.ResourceCPU])
		}
	}

	tcInt64, tcOk := tc.AsInt64()
	scInt64, scOk := sc.AsInt64()
	if !tcOk || !scOk {
		return nil, fmt.Errorf("unable to compute integer values of schedulable cores in the cluster")
	}
	clusterStatus.TotalCores = int32(tcInt64)
	clusterStatus.SchedulableCores = int32(scInt64)
	k.clusterStatus = clusterStatus
	return clusterStatus, nil
}

pkg/autoscaler/autoscaler_server.go:124
func (s *AutoScaler) syncConfigWithServer() (*apiv1.ConfigMap, error) {
	// Fetch autoscaler ConfigMap data from apiserver
	configMap, err := s.k8sClient.FetchConfigMap(s.k8sClient.GetNamespace(), s.configMapName)
	if err == nil {
		return configMap, nil
	}
	if s.defaultParams == nil {
		return nil, err
	}
	glog.V(0).Infof("ConfigMap not found: %v, will create one with default params", err)
	configMap, err = s.k8sClient.CreateConfigMap(s.k8sClient.GetNamespace(), s.configMapName, s.defaultParams)
	if err != nil {
		return nil, err
	}
	return configMap, nil
}

pkg/autoscaler/controller/plugin/plugin.go:32

// EnsureController ensures controller type and scaling params
func EnsureController(cont controller.Controller, configMap *apiv1.ConfigMap) (controller.Controller, error) {
	// Expect only one entry, which uses the name of control mode as the key
	if len(configMap.Data) != 1 {
		return nil, fmt.Errorf("invalid configMap format, expected only one entry, got: %v", configMap.Data)
	}
	for mode := range configMap.Data {
		// No need to reset controller if control pattern doesn't change
		if cont != nil && mode == cont.GetControllerType() {
			break
		}
		switch mode {
		case laddercontroller.ControllerType:
			cont = laddercontroller.NewLadderController()
		case linearcontroller.ControllerType:
			cont = linearcontroller.NewLinearController()
		default:
			return nil, fmt.Errorf("not a supported control mode: %v", mode)
		}
		glog.V(1).Infof("Set control mode to %v", mode)
	}

	// Sync config with controller
	if err := cont.SyncConfig(configMap); err != nil {
		return nil, fmt.Errorf("Error syncing configMap with controller: %v", err)
	}
	return cont, nil
}

pkg/autoscaler/k8sclient/k8sclient.go:172
func (k *k8sClient) UpdateReplicas(expReplicas int32) (prevRelicas int32, err error) {
	scale, err := k.clientset.Extensions().Scales(k.target.namespace).Get(k.target.kind, k.target.name)
	if err != nil {
		return 0, err
	}
	prevRelicas = scale.Spec.Replicas
	if expReplicas != prevRelicas {
		glog.V(0).Infof("Cluster status: SchedulableNodes[%v], SchedulableCores[%v]", k.clusterStatus.SchedulableNodes, k.clusterStatus.SchedulableCores)
		glog.V(0).Infof("Replicas are not as expected : updating replicas from %d to %d", prevRelicas, expReplicas)
		scale.Spec.Replicas = expReplicas
		_, err = k.clientset.Extensions().Scales(k.target.namespace).Update(k.target.kind, scale)
		if err != nil {
			return 0, err
		}
	}
	return prevRelicas, nil
}

pkg/autoscaler/controller/linearcontroller/linear_controller.go:50
type linearParams struct {
	CoresPerReplica           float64 `json:"coresPerReplica"`
	NodesPerReplica           float64 `json:"nodesPerReplica"`
	Min                       int     `json:"min"`
	Max                       int     `json:"max"`
	PreventSinglePointFailure bool    `json:"preventSinglePointFailure"`
}

kind: ConfigMap
apiVersion: v1
metadata:
  name: nginx-autoscaler
  namespace: default
data:
  linear: |-
    { 
      "coresPerReplica": 2,
      "nodesPerReplica": 1,
      "preventSinglePointFailure": true,
      "min": 1,
      "max": 100
    }

pkg/autoscaler/controller/linearcontroller/linear_controller.go:101

func (c *LinearController) GetExpectedReplicas(status *k8sclient.ClusterStatus) (int32, error) {
	// Get the expected replicas for the currently schedulable nodes and cores
	expReplicas := int32(c.getExpectedReplicasFromParams(int(status.SchedulableNodes), int(status.SchedulableCores)))

	return expReplicas, nil
}

func (c *LinearController) getExpectedReplicasFromParams(schedulableNodes, schedulableCores int) int {
	replicasFromCore := c.getExpectedReplicasFromParam(schedulableCores, c.params.CoresPerReplica)
	replicasFromNode := c.getExpectedReplicasFromParam(schedulableNodes, c.params.NodesPerReplica)
	// Prevent single point of failure by having at least 2 replicas when
	// there are more than one node.
	if c.params.PreventSinglePointFailure &&
		schedulableNodes > 1 &&
		replicasFromNode < 2 {
		replicasFromNode = 2
	}

	// Returns the results which yields the most replicas
	if replicasFromCore > replicasFromNode {
		return replicasFromCore
	}
	return replicasFromNode
}

func (c *LinearController) getExpectedReplicasFromParam(schedulableResources int, resourcesPerReplica float64) int {
	if resourcesPerReplica == 0 {
		return 1
	}
	res := math.Ceil(float64(schedulableResources) / resourcesPerReplica)
	if c.params.Max != 0 {
		res = math.Min(float64(c.params.Max), res)
	}
	return int(math.Max(float64(c.params.Min), res))
}

replicas = max( ceil( cores * 1/coresPerReplica ) , ceil( nodes * 1/nodesPerReplica ) )
replicas = min(replicas, max)
replicas = max(replicas, min)

pkg/autoscaler/controller/laddercontroller/ladder_controller.go:66
type paramEntry [2]int
type paramEntries []paramEntry
type ladderParams struct {
	CoresToReplicas paramEntries `json:"coresToReplicas"`
	NodesToReplicas paramEntries `json:"nodesToReplicas"`
}

kind: ConfigMap
apiVersion: v1
metadata:
  name: nginx-autoscaler
  namespace: default
data:
  ladder: |-
    { 
      "coresToReplicas":
      [
        [ 1,1 ],
        [ 3,3 ],
        [256,4],
        [ 512,5 ],
        [ 1024,7 ]
      ],
      "nodesToReplicas":
      [
        [ 1,1 ],
        [ 2,2 ],
        [100, 5],
        [200, 12]
      ]
    }

func (c *LadderController) GetExpectedReplicas(status *k8sclient.ClusterStatus) (int32, error) {
	// Get the expected replicas for the currently schedulable nodes and cores
	expReplicas := int32(c.getExpectedReplicasFromParams(int(status.SchedulableNodes), int(status.SchedulableCores)))

	return expReplicas, nil
}

func (c *LadderController) getExpectedReplicasFromParams(schedulableNodes, schedulableCores int) int {
	replicasFromCore := getExpectedReplicasFromEntries(schedulableCores, c.params.CoresToReplicas)
	replicasFromNode := getExpectedReplicasFromEntries(schedulableNodes, c.params.NodesToReplicas)

	// Returns the results which yields the most replicas
	if replicasFromCore > replicasFromNode {
		return replicasFromCore
	}
	return replicasFromNode
}

func getExpectedReplicasFromEntries(schedulableResources int, entries []paramEntry) int {
	if len(entries) == 0 {
		return 1
	}
	// Binary search for the corresponding replicas number
	pos := sort.Search(
		len(entries),
		func(i int) bool {
			return schedulableResources < entries[i][0]
		})
	if pos > 0 {
		pos = pos - 1
	}
	return entries[pos][1]
}

kind: ConfigMap
apiVersion: v1
metadata:
  name: kube-dns-autoscaler
  namespace: kube-system
data:
  linear: |
    {
    "nodesPerReplica": 10,
    "min": 1,
    "max": 50,
    "preventSinglePointFailure": true
    } 

‐‐‐

apiVersion: extensions/v1beta1
kind: Deployment
metadata:
  name: kube-dns-autoscaler
  namespace: kube-system
spec:
  template:
    metadata:
      labels:
        k8s-app: kube-dns-autoscaler
    spec:
      imagePullSecrets:
      - name: harborsecret
      containers:
      - name: autoscaler
        image: registry.vivo.xyz:4443/bigdata_release/cluster_proportional_autoscaler_amd64:1.0.0
        resources:
          requests:
            cpu: "50m"
            memory: "100Mi"
        command:
        - /cluster-proportional-autoscaler
        - --namespace=kube-system
        - --configmap=kube-dns-autoscaler
        - --target=Deployment/kube-dns
        - --default-params={"linear":{"nodesPerReplica":10,"min":1}}
        - --logtostderr=true
        - --v=2