opensource/cloudpods
3
0
Fork 0
mirror of https://github.com/yunionio/cloudpods.git synced 2026-09-01 15:07:17 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(scheduler): remove unused k8s extender

Browse Source
This commit is contained in:
Zexi Li
2020-12-19 21:44:57 +08:00
parent 0d67fcb4ce
commit e88861f2c4
8 changed files with 2 additions and 525 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/scheduler/algorithm/predicates/k8s/doc.go
-15
View File
@@ -1,15 +0,0 @@
// Copyright 2019 Yunion
//
// 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 k8s // import "yunion.io/x/onecloud/pkg/scheduler/algorithm/predicates/k8s"
pkg/scheduler/algorithm/predicates/k8s/host_status_predicate.go
-50
View File
@@ -1,50 +0,0 @@
// Copyright 2019 Yunion
//
// 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 k8s
import (
"fmt"
v1 "k8s.io/api/core/v1"
"k8s.io/client-go/kubernetes"
"yunion.io/x/onecloud/pkg/scheduler/algorithm/predicates/guest"
"yunion.io/x/onecloud/pkg/scheduler/cache/candidate"
)
type HostStatusPredicate struct{}
func (p *HostStatusPredicate) Clone() IPredicate {
return &HostStatusPredicate{}
}
func (p *HostStatusPredicate) Name() string {
return "host_status"
}
func (p *HostStatusPredicate) PreExecute(cli *kubernetes.Clientset, pod *v1.Pod, node *v1.Node, host *candidate.HostDesc) bool {
return true
}
func (p *HostStatusPredicate) Execute(cli *kubernetes.Clientset, pod *v1.Pod, node *v1.Node, host *candidate.HostDesc) (bool, error) {
if host.Status != guest.ExpectedStatus {
return false, fmt.Errorf("Host status is %s", host.Status)
}
if !host.GetEnabled() {
return false, fmt.Errorf("Host is disabled")
}
return true, nil
}
pkg/scheduler/algorithm/predicates/k8s/k8s.go
-97
View File
@@ -1,97 +0,0 @@
// Copyright 2019 Yunion
//
// 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 k8s
import (
"fmt"
v1 "k8s.io/api/core/v1"
"k8s.io/client-go/kubernetes"
"yunion.io/x/onecloud/pkg/scheduler/cache/candidate"
)
var PredicatesManager *SPredicatesManager
func init() {
PredicatesManager = newPredicatesManager()
PredicatesManager.Register(
&HostStatusPredicate{},
&NetworkPredicate{},
&LocalVolumePredicate{},
)
}
type IPredicate interface {
Name() string
Clone() IPredicate
PreExecute(cli *kubernetes.Clientset, pod *v1.Pod, node *v1.Node, host *candidate.HostDesc) bool
Execute(cli *kubernetes.Clientset, pod *v1.Pod, node *v1.Node, host *candidate.HostDesc) (bool, error)
}
type SPredicatesManager struct {
predicates []IPredicate
}
func newPredicatesManager() *SPredicatesManager {
man := &SPredicatesManager{
predicates: make([]IPredicate, 0),
}
return man
}
func (man *SPredicatesManager) Register(pres ...IPredicate) *SPredicatesManager {
for _, pre := range pres {
if !man.Has(pre) {
man.predicates = append(man.predicates, pre)
}
}
return man
}
func (man *SPredicatesManager) Has(newPre IPredicate) bool {
if len(man.predicates) == 0 {
return false
}
for _, pre := range man.predicates {
if pre.Name() == newPre.Name() {
return true
}
}
return false
}
func (man *SPredicatesManager) DoFilter(
k8sCli *kubernetes.Clientset,
pod *v1.Pod,
node *v1.Node,
host *candidate.HostDesc,
) (bool, error) {
for _, pre := range man.predicates {
tmpPre := pre.Clone()
if !tmpPre.PreExecute(k8sCli, pod, node, host) {
continue
}
fit, err := tmpPre.Execute(k8sCli, pod, node, host)
if err != nil {
return false, err
}
if !fit {
return false, fmt.Errorf("Filtered by %s", tmpPre.Name())
}
}
return true, nil
}
pkg/scheduler/algorithm/predicates/k8s/network_predicate.go
-124
View File
@@ -1,124 +0,0 @@
// Copyright 2019 Yunion
//
// 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 k8s
import (
"fmt"
v1 "k8s.io/api/core/v1"
"k8s.io/client-go/kubernetes"
"yunion.io/x/pkg/util/errors"
"yunion.io/x/pkg/util/netutils"
"yunion.io/x/onecloud/pkg/compute/models"
"yunion.io/x/onecloud/pkg/scheduler/api"
"yunion.io/x/onecloud/pkg/scheduler/cache/candidate"
)
const (
// k8s annotations for create pod
YUNION_CNI_NETWORK_ANNOTATION = "cni.yunion.io/network"
YUNION_CNI_IPADDR_ANNOTATION = "cni.yunion.io/ip"
)
type NetworkPredicate struct {
network string
ipAddr string
}
func (p *NetworkPredicate) Clone() IPredicate {
return &NetworkPredicate{}
}
func (p *NetworkPredicate) Name() string {
return "network"
}
func (p *NetworkPredicate) PreExecute(cli *kubernetes.Clientset, pod *v1.Pod, node *v1.Node, host *candidate.HostDesc) bool {
net, netCont := pod.Annotations[YUNION_CNI_NETWORK_ANNOTATION]
ipAddr, ipCont := pod.Annotations[YUNION_CNI_IPADDR_ANNOTATION]
p.network = net
p.ipAddr = ipAddr
return netCont || ipCont
}
func (p *NetworkPredicate) Execute(cli *kubernetes.Clientset, pod *v1.Pod, node *v1.Node, host *candidate.HostDesc) (bool, error) {
hostNets := host.Networks
if p.network != "" {
err := p.checkByNetworks(hostNets)
if err != nil {
return false, err
}
}
if p.ipAddr != "" {
err := p.checkNetworksIP(p.ipAddr, hostNets)
if err != nil {
return false, err
}
}
return true, nil
}
func (p NetworkPredicate) checkByNetworks(nets []*api.CandidateNetwork) error {
if len(nets) == 0 {
return fmt.Errorf("Network is empty")
}
errs := make([]error, 0)
for _, net := range nets {
err := p.checkByNetwork(net.SNetwork)
if err == nil {
return nil
}
errs = append(errs, err)
}
return errors.NewAggregate(errs)
}
func (p NetworkPredicate) checkByNetwork(net *models.SNetwork) error {
if net.GetPorts() <= 0 {
return fmt.Errorf("Network %s no free IPs", net.Name)
}
if !(p.network == net.Name || p.network == net.Id) {
return fmt.Errorf("Network %s:%s or id not match %s", net.Name, net.Id, p.network)
}
return nil
}
func (p NetworkPredicate) checkNetworksIP(ip string, nets []*api.CandidateNetwork) error {
if len(nets) == 0 {
return fmt.Errorf("Network is empty")
}
errs := make([]error, 0)
for _, net := range nets {
err := p.checkNetworkIP(ip, net.SNetwork)
if err == nil {
return nil
}
errs = append(errs, err)
}
return errors.NewAggregate(errs)
}
func (p NetworkPredicate) checkNetworkIP(ip string, net *models.SNetwork) error {
ipAddr, err := netutils.NewIPV4Addr(ip)
if err != nil {
return err
}
if ok := net.GetIPRange().Contains(ipAddr); !ok {
return fmt.Errorf("Network %s not contains ip %s", net.Name, ip)
}
return nil
}
pkg/scheduler/algorithm/predicates/k8s/volume_predicate.go
-103
View File
@@ -1,103 +0,0 @@
// Copyright 2019 Yunion
//
// 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 k8s
import (
"fmt"
v1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/kubernetes"
"yunion.io/x/log"
"yunion.io/x/onecloud/pkg/scheduler/cache/candidate"
)
const (
YUNION_CSI_STORAGECLASS = "csi-yunion"
)
type LocalVolumePredicate struct {
pvcs []*v1.PersistentVolumeClaim
}
func (p *LocalVolumePredicate) Clone() IPredicate {
return &LocalVolumePredicate{}
}
func (p *LocalVolumePredicate) Name() string {
return "local-volume"
}
func (p *LocalVolumePredicate) PreExecute(cli *kubernetes.Clientset, pod *v1.Pod, node *v1.Node, host *candidate.HostDesc) bool {
if cli == nil {
log.Errorf("k8s client is nil, not execute %s filter", p.Name())
return false
}
pvcs := make([]*v1.PersistentVolumeClaim, 0)
if pod.Spec.Volumes != nil && len(pod.Spec.Volumes) > 0 {
for _, v := range pod.Spec.Volumes {
if v.PersistentVolumeClaim == nil {
continue
}
pvcName := v.PersistentVolumeClaim.ClaimName
pvc, err := cli.CoreV1().PersistentVolumeClaims(pod.Namespace).Get(pvcName, metav1.GetOptions{})
if err != nil {
log.Warningf("Not found pvc %s for pod %s/%s", pvcName, pod.Namespace, pod.Name)
return false
}
// pvc's StorageClassName must "csi-yunion"
if pvc.Spec.StorageClassName != nil && *(pvc.Spec.StorageClassName) != YUNION_CSI_STORAGECLASS {
continue
}
pvcs = append(pvcs, pvc)
}
}
p.pvcs = pvcs
return len(pvcs) != 0
}
func (p *LocalVolumePredicate) getPvcRequestSize(pvc *v1.PersistentVolumeClaim) int64 {
req := pvc.Spec.Resources.Requests[v1.ResourceStorage]
return req.Value()
}
func (p *LocalVolumePredicate) Execute(cli *kubernetes.Clientset, pod *v1.Pod, node *v1.Node, host *candidate.HostDesc) (bool, error) {
var reqSize int64
for _, pvc := range p.pvcs {
pvName := pvc.Spec.VolumeName
if pvName != "" {
pv, _ := cli.CoreV1().PersistentVolumes().Get(pvName, metav1.GetOptions{})
if pv != nil {
// PersistentVolume already exists
log.V(10).Debugf("PV %s already exists", pv)
continue
}
}
reqSize += p.getPvcRequestSize(pvc)
}
reqSizeMB := reqSize / 1024 / 1024
return p.canHostStorageCreateVol(host, reqSizeMB)
}
func (p *LocalVolumePredicate) canHostStorageCreateVol(host *candidate.HostDesc, reqSize int64) (bool, error) {
freeSize := host.GetFreeStorageSizeOfType("local", false)
log.Debugf("[host %s] PVC request %dMB, free %dMB", host.Name, reqSize, freeSize)
if freeSize > reqSize {
return true, nil
}
return false, fmt.Errorf("Out of local storage for volume: %d/%d (request/free)MB", reqSize, freeSize)
}
pkg/scheduler/handler/handler.go
-1
View File
@@ -41,7 +41,6 @@ func InstallHandler(r *gin.Engine) {
r.POST("/scheduler/:action/:ident", timer(schedulerActionIdentHandler))
InstallPingHandler(r)
InstallVersionHandler(r)
InstallK8sSchedExtenderHandler(r)
}
func timer(f gin.HandlerFunc) gin.HandlerFunc {
pkg/scheduler/handler/k8s_extender.go
-123
View File
@@ -1,123 +0,0 @@
// Copyright 2019 Yunion
//
// 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 handler
import (
"encoding/json"
"fmt"
"net/http"
gin "github.com/gin-gonic/gin"
v1 "k8s.io/api/core/v1"
schedulerapi "k8s.io/kubernetes/pkg/scheduler/api/v1"
"yunion.io/x/log"
k8spredicates "yunion.io/x/onecloud/pkg/scheduler/algorithm/predicates/k8s"
schedman "yunion.io/x/onecloud/pkg/scheduler/manager"
)
func InstallK8sSchedExtenderHandler(r *gin.Engine) {
r.POST("/k8s/predicates", timer(k8sPredicatesHandler))
//r.POST("/k8s/priorities", timer(k8sPrioritizeHandler))
}
func k8sPredicatesHandler(c *gin.Context) {
if !schedman.IsReady() {
c.AbortWithError(http.StatusBadRequest, fmt.Errorf("Global scheduler not init"))
return
}
var extenderArgs schedulerapi.ExtenderArgs
var extenderFilterResult *schedulerapi.ExtenderFilterResult
if err := json.NewDecoder(c.Request.Body).Decode(&extenderArgs); err != nil {
extenderFilterResult = &schedulerapi.ExtenderFilterResult{
Nodes: nil,
FailedNodes: nil,
Error: err.Error(),
}
} else {
extenderFilterResult = k8sPredicateFunc(extenderArgs)
}
c.JSON(http.StatusOK, extenderFilterResult)
}
func k8sPredicateFunc(args schedulerapi.ExtenderArgs) *schedulerapi.ExtenderFilterResult {
pod := args.Pod
canSchedule := make([]v1.Node, 0, len(args.Nodes.Items))
canNotSchedule := make(map[string]string)
for _, node := range args.Nodes.Items {
result, err := doK8sPredicates(pod, &node)
if err != nil {
canNotSchedule[node.Name] = fmt.Sprintf("%s: %v", node.Name, err)
} else {
if result {
canSchedule = append(canSchedule, node)
}
}
}
return &schedulerapi.ExtenderFilterResult{
Nodes: &v1.NodeList{
Items: canSchedule,
},
FailedNodes: canNotSchedule,
Error: "",
}
}
func doK8sPredicates(pod *v1.Pod, node *v1.Node) (bool, error) {
hosts, err := schedman.GetK8sCandidateHosts(node.Name)
if err != nil {
return false, err
}
if len(hosts) == 0 {
return false, fmt.Errorf("Not found candidate host %s", node.Name)
}
k8sCli, err := schedman.GetK8sClient()
if err != nil {
log.Warningf("Get k8s client error: %v, some predicates will not execute", err)
}
return k8spredicates.PredicatesManager.DoFilter(k8sCli, pod, node, hosts[0])
}
func k8sPrioritizeHandler(c *gin.Context) {
var extenderArgs schedulerapi.ExtenderArgs
var hostPriorityList *schedulerapi.HostPriorityList
if err := json.NewDecoder(c.Request.Body).Decode(&extenderArgs); err != nil {
c.AbortWithError(http.StatusBadRequest, fmt.Errorf("Decode json body: %v", err))
return
}
hostPriorityList, err := k8sPrioritizeFunc(extenderArgs)
if err != nil {
c.AbortWithError(http.StatusInternalServerError, fmt.Errorf("Prioritize error: %v", err))
return
}
c.JSON(http.StatusOK, hostPriorityList)
}
// TODO: support priority extender
func k8sPrioritizeFunc(args schedulerapi.ExtenderArgs) (*schedulerapi.HostPriorityList, error) {
nodes := args.Nodes.Items
var priorityList schedulerapi.HostPriorityList = make([]schedulerapi.HostPriority, 0)
for _, node := range nodes {
priorityList = append(priorityList, schedulerapi.HostPriority{
Host: node.Name,
Score: 0,
})
}
return &priorityList, nil
}
pkg/scheduler/manager/manager.go
+2 -12
View File
@@ -19,8 +19,6 @@ import (
"fmt"
"time"
"k8s.io/client-go/kubernetes"
"yunion.io/x/log"
"yunion.io/x/onecloud/pkg/scheduler/api"
@@ -29,8 +27,6 @@ import (
"yunion.io/x/onecloud/pkg/scheduler/core"
"yunion.io/x/onecloud/pkg/scheduler/data_manager"
schedmodels "yunion.io/x/onecloud/pkg/scheduler/models"
o "yunion.io/x/onecloud/pkg/scheduler/options"
"yunion.io/x/onecloud/pkg/util/k8s"
)
const defaultIgnorePool = true
@@ -43,9 +39,8 @@ type SchedulerManager struct {
HistoryManager *HistoryManager
TaskManager *TaskManager
DataManager *data_manager.DataManager
CandidateManager *data_manager.CandidateManager
KubeClusterManager *k8s.SKubeClusterManager
DataManager *data_manager.DataManager
CandidateManager *data_manager.CandidateManager
}
func NewSchedulerManager(stopCh <-chan struct{}) *SchedulerManager {
@@ -56,7 +51,6 @@ func NewSchedulerManager(stopCh <-chan struct{}) *SchedulerManager {
sm.CompletedManager = NewCompletedManager(stopCh)
sm.HistoryManager = NewHistoryManager(stopCh)
sm.TaskManager = NewTaskManager(stopCh)
sm.KubeClusterManager = k8s.NewKubeClusterManager(o.GetOptions().Region, 30*time.Second)
return sm
}
@@ -65,10 +59,6 @@ func GetScheduleManager() *SchedulerManager {
return schedManager
}
func GetK8sClient() (*kubernetes.Clientset, error) {
return GetScheduleManager().KubeClusterManager.GetK8sClient()
}
func InitAndStart(stopCh <-chan struct{}) {
if schedManager != nil {
log.Warningf("Global scheduler already init.")