/*
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 scheduler
import (
"bytes"
"encoding/json"
"fmt"
"net/http"
"strings"
"time"
v1 "k8s.io/api/core/v1"
utilnet "k8s.io/apimachinery/pkg/util/net"
"k8s.io/apimachinery/pkg/util/sets"
restclient "k8s.io/client-go/rest"
extenderv1 "k8s.io/kube-scheduler/extender/v1"
schedulerapi "k8s.io/kubernetes/pkg/scheduler/apis/config"
"k8s.io/kubernetes/pkg/scheduler/framework"
)
const (
// DefaultExtenderTimeout defines the default extender timeout in second.
DefaultExtenderTimeout = 5 * time.Second
)
// HTTPExtender implements the Extender interface.
type HTTPExtender struct {
extenderURL string
preemptVerb string
filterVerb string
prioritizeVerb string
bindVerb string
weight int64
client *http.Client
nodeCacheCapable bool
managedResources sets.Set[string]
ignorable bool
}
//这段代码定义了一个名为HTTPExtender的结构体，它实现了Extender接口。
//HTTPExtender表示一个HTTP扩展器，用于扩展Kubernetes调度器的功能。
//它通过发送HTTP请求与Kubernetes调度器进行交互，以实现自定义的调度逻辑。
//该结构体包含以下字段： - extenderURL：扩展器的URL，用于向扩展器发送HTTP请求。
//- preemptVerb：抢占操作的动词，表示扩展器在抢占过程中执行的操作。
//- filterVerb：过滤操作的动词，表示扩展器在过滤节点时执行的操作。
//- prioritizeVerb：优先级排序操作的动词，表示扩展器在对节点进行优先级排序时执行的操作。
//- bindVerb：绑定操作的动词，表示扩展器在绑定Pod到节点时执行的操作。
//- weight：扩展器的权重，用于在优先级排序时对多个扩展器的结果进行加权计算。
//- client：用于发送HTTP请求的HTTP客户端。
//- nodeCacheCapable：表示扩展器是否支持节点缓存。
//如果为true，调度器会将节点信息缓存在本地，以减少对扩展器的请求次数。
//- managedResources：表示扩展器管理的资源集合。
//调度器会将这些资源的调度委托给扩展器处理。
//- ignorable：表示扩展器是否可以被忽略。如果为true，当扩展器出现错误时，调度器可以忽略该错误并继续进行调度。
//此外，代码还定义了一个常量DefaultExtenderTimeout，它表示默认的扩展器超时时间，单位为秒。
func makeTransport(config *schedulerapi.Extender) (http.RoundTripper, error) {
var cfg restclient.Config
if config.TLSConfig != nil {
cfg.TLSClientConfig.Insecure = config.TLSConfig.Insecure
cfg.TLSClientConfig.ServerName = config.TLSConfig.ServerName
cfg.TLSClientConfig.CertFile = config.TLSConfig.CertFile
cfg.TLSClientConfig.KeyFile = config.TLSConfig.KeyFile
cfg.TLSClientConfig.CAFile = config.TLSConfig.CAFile
cfg.TLSClientConfig.CertData = config.TLSConfig.CertData
cfg.TLSClientConfig.KeyData = config.TLSConfig.KeyData
cfg.TLSClientConfig.CAData = config.TLSConfig.CAData
}
if config.EnableHTTPS {
hasCA := len(cfg.CAFile) > 0 || len(cfg.CAData) > 0
if !hasCA {
cfg.Insecure = true
}
}
tlsConfig, err := restclient.TLSConfigFor(&cfg)
if err != nil {
return nil, err
}
if tlsConfig != nil {
return utilnet.SetTransportDefaults(&http.Transport{
TLSClientConfig: tlsConfig,
}), nil
}
return utilnet.SetTransportDefaults(&http.Transport{}), nil
}
//该函数根据给定的schedulerapi.Extender配置创建一个http.RoundTripper传输对象。
//它首先将config.TLSConfig中的配置项复制到cfg.TLSClientConfig中，然后根据config.EnableHTTPS的值设置cfg.Insecure。
//接下来，它通过调用restclient.TLSConfigFor(&cfg)来获取TLS配置，并将其设置到http.Transport中。
//最后，返回设置了默认值的http.Transport对象。如果在过程中出现错误，则返回错误。
// NewHTTPExtender creates an HTTPExtender object.
func NewHTTPExtender(config *schedulerapi.Extender) (framework.Extender, error) {
if config.HTTPTimeout.Duration.Nanoseconds() == 0 {
config.HTTPTimeout.Duration = time.Duration(DefaultExtenderTimeout)
}
transport, err := makeTransport(config)
if err != nil {
return nil, err
}
client := &http.Client{
Transport: transport,
Timeout: config.HTTPTimeout.Duration,
}
managedResources := sets.New[string]()
for _, r := range config.ManagedResources {
managedResources.Insert(string(r.Name))
}
return &HTTPExtender{
extenderURL: config.URLPrefix,
preemptVerb: config.PreemptVerb,
filterVerb: config.FilterVerb,
prioritizeVerb: config.PrioritizeVerb,
bindVerb: config.BindVerb,
weight: config.Weight,
client: client,
nodeCacheCapable: config.NodeCacheCapable,
managedResources: managedResources,
ignorable: config.Ignorable,
}, nil
}
//该函数用于创建一个HTTPExtender对象。它根据传入的config参数初始化HTTPExtender对象的属性，并返回该对象。
//其中，makeTransport函数用于创建一个http.Transport对象，http.Client对象则使用该Transport对象以及config参数中的HTTP超时时间进行初始化。
//此外，该函数还对config.ManagedResources进行遍历，将其中的Name字段转换为string类型，并插入到managedResources集合中。
//最后，该函数返回一个初始化完成的HTTPExtender对象。
// Name returns extenderURL to identify the extender.
func (h *HTTPExtender) Name() string {
return h.extenderURL
}
// IsIgnorable returns true indicates scheduling should not fail when this extender
// is unavailable
func (h *HTTPExtender) IsIgnorable() bool {
return h.ignorable
}
// SupportsPreemption returns true if an extender supports preemption.
// An extender should have preempt verb defined and enabled its own node cache.
func (h *HTTPExtender) SupportsPreemption() bool {
return len(h.preemptVerb) > 0
}
//这段代码定义了一个名为HTTPExtender的结构体及其三个方法。
//1. Name()方法返回extenderURL，用于标识扩展程序。
//2. IsIgnorable()方法返回一个布尔值，指示当此扩展程序不可用时，调度是否不应失败。如果ignorable字段为true，则返回true。
//3. SupportsPreemption()方法返回一个布尔值，表示扩展程序是否支持抢占。
//如果preemptVerb字段的长度大于0，则返回true。
//这意味着扩展程序应该定义抢占动词并启用自己的节点缓存。
// ProcessPreemption returns filtered candidate nodes and victims after running preemption logic in extender.
func (h *HTTPExtender) ProcessPreemption(
pod *v1.Pod,
nodeNameToVictims map[string]*extenderv1.Victims,
nodeInfos framework.NodeInfoLister,
) (map[string]*extenderv1.Victims, error) {
var (
result extenderv1.ExtenderPreemptionResult
args *extenderv1.ExtenderPreemptionArgs
)
if !h.SupportsPreemption() {
return nil, fmt.Errorf("preempt verb is not defined for extender %v but run into ProcessPreemption", h.extenderURL)
}
if h.nodeCacheCapable {
// If extender has cached node info, pass NodeNameToMetaVictims in args.
nodeNameToMetaVictims := convertToMetaVictims(nodeNameToVictims)
args = &extenderv1.ExtenderPreemptionArgs{
Pod: pod,
NodeNameToMetaVictims: nodeNameToMetaVictims,
}
} else {
args = &extenderv1.ExtenderPreemptionArgs{
Pod: pod,
NodeNameToVictims: nodeNameToVictims,
}
}
if err := h.send(h.preemptVerb, args, &result); err != nil {
return nil, err
}
// Extender will always return NodeNameToMetaVictims.
// So let's convert it to NodeNameToVictims by using <nodeInfos>.
newNodeNameToVictims, err := h.convertToVictims(result.NodeNameToMetaVictims, nodeInfos)
if err != nil {
return nil, err
}
// Do not override <nodeNameToVictims>.
return newNodeNameToVictims, nil
}
//该函数是用于处理抢占逻辑的HTTP扩展器方法。
//它根据传入的Pod和节点信息，通过调用扩展器的抢占逻辑，返回过滤后的候选节点和受害者。
//- 首先，函数检查扩展器是否支持抢占。如果不支持，则返回错误。
//- 接下来，根据扩展器是否具有缓存的节点信息，构建不同的ExtenderPreemptionArgs对象，其中包含Pod和节点信息。
//- 然后，通过调用扩展器的指定动词（preempt），将ExtenderPreemptionArgs发送给扩展器，并将结果存储在ExtenderPreemptionResult中。
//- 最后，将扩展器返回的NodeNameToMetaVictims转换为NodeNameToVictims，并返回新的NodeNameToVictims对象。
//该函数返回的地图map[string]*extenderv1.Victims表示了每个节点的受害者列表，其中每个受害者包含了被抢占的Pod列表。
// convertToVictims converts "nodeNameToMetaVictims" from object identifiers,
// such as UIDs and names, to object pointers.
func (h *HTTPExtender) convertToVictims(
nodeNameToMetaVictims map[string]*extenderv1.MetaVictims,
nodeInfos framework.NodeInfoLister,
) (map[string]*extenderv1.Victims, error) {
nodeNameToVictims := map[string]*extenderv1.Victims{}
for nodeName, metaVictims := range nodeNameToMetaVictims {
nodeInfo, err := nodeInfos.Get(nodeName)
if err != nil {
return nil, err
}
victims := &extenderv1.Victims{
Pods: []*v1.Pod{},
NumPDBViolations: metaVictims.NumPDBViolations,
}
for _, metaPod := range metaVictims.Pods {
pod, err := h.convertPodUIDToPod(metaPod, nodeInfo)
if err != nil {
return nil, err
}
victims.Pods = append(victims.Pods, pod)
}
nodeNameToVictims[nodeName] = victims
}
return nodeNameToVictims, nil
}
//该函数是一个Go语言函数，它将从对象标识符（如UID和名称）组成的"nodeNameToMetaVictims"映射转换为对象指针组成的"nodeNameToVictims"映射。
//该函数使用给定的节点信息获取节点名称对应的受害者信息，并将其转换为包含Pod信息和PDB违规数量的Victims对象。
//具体来说，它遍历输入的映射，获取每个节点的MetaVictims对象，并通过节点信息获取节点的Pod信息。
//然后，它将MetaPod对象转换为Pod对象，并将其添加到Victims对象的Pod列表中。
//最后，它将转换后的Victims对象添加到输出映射中，并返回该映射。如果在转换过程中发生错误，函数将返回错误。
// convertPodUIDToPod returns v1.Pod object for given MetaPod and node info.
// The v1.Pod object is restored by nodeInfo.Pods().
// It returns an error if there's cache inconsistency between default scheduler
// and extender, i.e. when the pod is not found in nodeInfo.Pods.
func (h *HTTPExtender) convertPodUIDToPod(
metaPod *extenderv1.MetaPod,
nodeInfo *framework.NodeInfo) (*v1.Pod, error) {
for _, p := range nodeInfo.Pods {
if string(p.Pod.UID) == metaPod.UID {
return p.Pod, nil
}
}
return nil, fmt.Errorf("extender: %v claims to preempt pod (UID: %v) on node: %v, but the pod is not found on that node",
h.extenderURL, metaPod, nodeInfo.Node().Name)
}
// convertToMetaVictims converts from struct type to meta types.
func convertToMetaVictims(
nodeNameToVictims map[string]*extenderv1.Victims,
) map[string]*extenderv1.MetaVictims {
nodeNameToMetaVictims := map[string]*extenderv1.MetaVictims{}
for node, victims := range nodeNameToVictims {
metaVictims := &extenderv1.MetaVictims{
Pods: []*extenderv1.MetaPod{},
NumPDBViolations: victims.NumPDBViolations,
}
for _, pod := range victims.Pods {
metaPod := &extenderv1.MetaPod{
UID: string(pod.UID),
}
metaVictims.Pods = append(metaVictims.Pods, metaPod)
}
nodeNameToMetaVictims[node] = metaVictims
}
return nodeNameToMetaVictims
}
//该函数的功能是将从结构体类型转换为元类型。
//它接收一个nodeNameToVictims参数，该参数是一个map，其中键是节点名称，值是*extenderv1.Victims类型的指针。
//函数创建一个空的nodeNameToMetaVictims映射，然后遍历nodeNameToVictims中的每个元素。
//对于每个节点，它创建一个新的extenderv1.MetaVictims实例，并将NumPDBViolations从旧的victims实例复制到新的metaVictims实例中。
//然后，它遍历旧的victims实例中的每个Pod，为每个Pod创建一个新的extenderv1.MetaPod实例，并将UID从旧的pod实例复制到新的metaPod实例中。
//最后，它将新的metaVictims实例添加到nodeNameToMetaVictims映射中，并返回该映射。
// Filter based on extender implemented predicate functions. The filtered list is
// expected to be a subset of the supplied list; otherwise the function returns an error.
// The failedNodes and failedAndUnresolvableNodes optionally contains the list
// of failed nodes and failure reasons, except nodes in the latter are
// unresolvable.
func (h *HTTPExtender) Filter(
pod *v1.Pod,
nodes []*framework.NodeInfo,
) (filteredList []*framework.NodeInfo, failedNodes, failedAndUnresolvableNodes extenderv1.FailedNodesMap, err error) {
var (
result extenderv1.ExtenderFilterResult
nodeList *v1.NodeList
nodeNames *[]string
nodeResult []*framework.NodeInfo
args *extenderv1.ExtenderArgs
)
fromNodeName := make(map[string]*framework.NodeInfo)
for _, n := range nodes {
fromNodeName[n.Node().Name] = n
}
if h.filterVerb == "" {
return nodes, extenderv1.FailedNodesMap{}, extenderv1.FailedNodesMap{}, nil
}
if h.nodeCacheCapable {
nodeNameSlice := make([]string, 0, len(nodes))
for _, node := range nodes {
nodeNameSlice = append(nodeNameSlice, node.Node().Name)
}
nodeNames = &nodeNameSlice
} else {
nodeList = &v1.NodeList{}
for _, node := range nodes {
nodeList.Items = append(nodeList.Items, *node.Node())
}
}
args = &extenderv1.ExtenderArgs{
Pod: pod,
Nodes: nodeList,
NodeNames: nodeNames,
}
if err := h.send(h.filterVerb, args, &result); err != nil {
return nil, nil, nil, err
}
if result.Error != "" {
return nil, nil, nil, fmt.Errorf(result.Error)
}
if h.nodeCacheCapable && result.NodeNames != nil {
nodeResult = make([]*framework.NodeInfo, len(*result.NodeNames))
for i, nodeName := range *result.NodeNames {
if n, ok := fromNodeName[nodeName]; ok {
nodeResult[i] = n
} else {
return nil, nil, nil, fmt.Errorf(
"extender %q claims a filtered node %q which is not found in the input node list",
h.extenderURL, nodeName)
}
}
} else if result.Nodes != nil {
nodeResult = make([]*framework.NodeInfo, len(result.Nodes.Items))
for i := range result.Nodes.Items {
nodeResult[i] = framework.NewNodeInfo()
nodeResult[i].SetNode(&result.Nodes.Items[i])
}
}
return nodeResult, result.FailedNodes, result.FailedAndUnresolvableNodes, nil
}
//该函数是一个过滤函数，基于扩展器实现的谓词函数对节点进行过滤。
//函数的输入是一个Pod对象和一个节点信息数组，输出是过滤后的节点信息数组、失败的节点信息映射、不可解析的节点信息映射和错误信息。
//函数首先通过遍历输入节点信息数组，创建一个从节点名称到节点信息的映射。
//然后根据扩展器的过滤动词，决定使用节点名称列表还是节点列表作为输入参数，构建ExtenderArgs对象，并调用扩展器的过滤方法。
//如果过滤方法返回错误，函数直接返回错误。如果过滤方法成功，函数根据返回结果构建过滤后的节点信息数组，并返回。
//如果扩展器声明了一个节点被过滤，但是在输入节点信息数组中找不到该节点，则函数返回错误。
//如果扩展器返回的结果中包含失败的节点和不可解析的节点信息，则将其添加到相应的映射中。
//总之，该函数通过调用扩展器的过滤方法，对节点进行过滤，并返回过滤后的节点信息数组和相关错误信息。
// Prioritize based on extender implemented priority functions. Weight*priority is added
// up for each such priority function. The returned score is added to the score computed
// by Kubernetes scheduler. The total score is used to do the host selection.
func (h *HTTPExtender) Prioritize(pod *v1.Pod, nodes []*framework.NodeInfo) (*extenderv1.HostPriorityList, int64, error) {
var (
result extenderv1.HostPriorityList
nodeList *v1.NodeList
nodeNames *[]string
args *extenderv1.ExtenderArgs
)
if h.prioritizeVerb == "" {
result := extenderv1.HostPriorityList{}
for _, node := range nodes {
result = append(result, extenderv1.HostPriority{Host: node.Node().Name, Score: 0})
}
return &result, 0, nil
}
if h.nodeCacheCapable {
nodeNameSlice := make([]string, 0, len(nodes))
for _, node := range nodes {
nodeNameSlice = append(nodeNameSlice, node.Node().Name)
}
nodeNames = &nodeNameSlice
} else {
nodeList = &v1.NodeList{}
for _, node := range nodes {
nodeList.Items = append(nodeList.Items, *node.Node())
}
}
args = &extenderv1.ExtenderArgs{
Pod: pod,
Nodes: nodeList,
NodeNames: nodeNames,
}
if err := h.send(h.prioritizeVerb, args, &result); err != nil {
return nil, 0, err
}
return &result, h.weight, nil
}
//该函数是一个Go语言函数，定义在HTTPExtender结构体中，用于根据扩展程序实现的优先级函数对节点进行优先级排序。
//该函数将权重乘以优先级的总和添加到每个此类优先级函数中。
//返回的分数将添加到Kubernetes调度程序计算的分数中。总
//分数用于进行主机选择。
//函数接受一个*v1.Pod类型的pod参数，一个[]*framework.NodeInfo类型的nodes参数，以及一个error类型的结果参数。
//函数首先定义了一些局部变量，包括一个extenderv1.HostPriorityList类型的result变量，
//一个*v1.NodeList类型的nodeList变量，一个*[]string类型的nodeNames变量，以及一个*extenderv1.ExtenderArgs类型的args变量。
//如果h.prioritizeVerb为空字符串，则将每个节点的分数设置为0，并返回结果。
//如果h.nodeCacheCapable为true，则将节点名称添加到nodeNameSlice切片中，并将其赋值给nodeNames变量。
//否则，将节点对象追加到nodeList的Items字段中。
//接下来，将pod、nodeList和nodeNames赋值给args的相应字段。
//最后，调用h.send方法，将h.prioritizeVerb、args和&result作为参数传递，并检查是否有错误发生。
//如果有错误，则返回错误。否则，返回结果和权重。
// Bind delegates the action of binding a pod to a node to the extender.
func (h *HTTPExtender) Bind(binding *v1.Binding) error {
var result extenderv1.ExtenderBindingResult
if !h.IsBinder() {
// This shouldn't happen as this extender wouldn't have become a Binder.
return fmt.Errorf("unexpected empty bindVerb in extender")
}
req := &extenderv1.ExtenderBindingArgs{
PodName: binding.Name,
PodNamespace: binding.Namespace,
PodUID: binding.UID,
Node: binding.Target.Name,
}
if err := h.send(h.bindVerb, req, &result); err != nil {
return err
}
if result.Error != "" {
return fmt.Errorf(result.Error)
}
return nil
}
// IsBinder returns whether this extender is configured for the Bind method.
func (h *HTTPExtender) IsBinder() bool {
return h.bindVerb != ""
}
// IsPrioritizer returns whether this extender is configured for the Prioritize method.
func (h *HTTPExtender) IsPrioritizer() bool {
return h.prioritizeVerb != ""
}
// IsFilter returns whether this extender is configured for the Filter method.
func (h *HTTPExtender) IsFilter() bool {
return h.filterVerb != ""
}
//这段代码定义了一个名为HTTPExtender的结构体及其相关方法。
//这个结构体用于委托将Pod绑定到节点的操作给扩展器。
//- Bind方法用于将Pod绑定到节点
//。它首先检查当前扩展器是否配置为绑定扩展器，如果不是，则返回错误。
//然后创建一个ExtenderBindingArgs请求，包含Pod的名称、命名空间、UID以及目标节点的名称，并通过调用send方法将请求发送给扩展器。
//如果发送请求或处理结果出现错误，则返回相应的错误。
//- IsBinder方法用于判断当前扩展器是否配置了绑定方法。如果bindVerb不为空，则表示配置了绑定方法。
//- IsPrioritizer方法用于判断当前扩展器是否配置了优先级方法。如果prioritizeVerb不为空，则表示配置了优先级方法。
//- IsFilter方法用于判断当前扩展器是否配置了过滤方法。如果filterVerb不为空，则表示配置了过滤方法。
// Helper function to send messages to the extender
func (h *HTTPExtender) send(action string, args interface{}, result interface{}) error {
out, err := json.Marshal(args)
if err != nil {
return err
}
url := strings.TrimRight(h.extenderURL, "/") + "/" + action
req, err := http.NewRequest("POST", url, bytes.NewReader(out))
if err != nil {
return err
}
req.Header.Set("Content-Type", "application/json")
resp, err := h.client.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("failed %v with extender at URL %v, code %v", action, url, resp.StatusCode)
}
return json.NewDecoder(resp.Body).Decode(result)
}
//该函数是一个发送消息给扩展程序的辅助函数。
//它使用HTTP POST请求将动作、参数和结果发送到指定的扩展程序URL。
//具体步骤如下：
//1. 将参数args转换为JSON格式。
//2. 构建请求URL，通过拼接扩展程序URL和动作字符串。
//3. 创建HTTP POST请求，设置请求头的Content-Type为application/json，并将JSON数据作为请求体。
//4. 发送请求并获取响应。
//5. 检查响应状态码，如果不是200 OK，则返回错误。
//6. 将响应体解码为结果参数result。 如果在上述过程中发生错误，则返回相应的错误。
// IsInterested returns true if at least one extended resource requested by
// this pod is managed by this extender.
func (h *HTTPExtender) IsInterested(pod *v1.Pod) bool {
if h.managedResources.Len() == 0 {
return true
}
if h.hasManagedResources(pod.Spec.Containers) {
return true
}
if h.hasManagedResources(pod.Spec.InitContainers) {
return true
}
return false
}
//该函数是一个Go语言函数，名为IsInterested，它属于HTTPExtender类型。
//函数用于判断给定的Pod是否至少请求了一个由该extender管理的扩展资源。
//函数返回一个布尔值，如果Pod请求了至少一个由该extender管理的扩展资源，则返回true，否则返回false。
//函数主要包含以下两个步骤：
//1. 首先，函数检查HTTPExtender的managedResources列表是否为空。如果为空，则表示该extender管理所有资源，因此直接返回true。
//2. 如果managedResources不为空，函数会分别检查Pod的Spec.Containers和Spec.InitContainers字段中是否包含由该extender管理的扩展资源。
//如果存在至少一个由extender管理的扩展资源，则返回true。如果两个字段中都不包含由extender管理的扩展资源，则返回false。
//总结：该函数用于判断给定的Pod是否请求了由该extender管理的扩展资源，根据管理资源列表和Pod的容器配置进行匹配判断。
func (h *HTTPExtender) hasManagedResources(containers []v1.Container) bool {
for i := range containers {
container := &containers[i]
for resourceName := range container.Resources.Requests {
if h.managedResources.Has(string(resourceName)) {
return true
}
}
for resourceName := range container.Resources.Limits {
if h.managedResources.Has(string(resourceName)) {
return true
}
}
}
return false
}
//该函数用于判断给定的容器列表中是否包含有管理资源。
//具体实现为遍历容器列表，再遍历容器的资源请求和限制，
//若存在管理资源则返回true，否则返回false。