custom-kube-scheduler/kubernetes-v1.15.4/pkg/scheduler/algorithm/priorities/custom_resource_allocation.go

/*
Copyright 2016 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 priorities

import (
	"database/sql"
	"fmt"
	"os"

	_ "github.com/go-sql-driver/mysql"
	client "github.com/influxdata/influxdb1-client/v2"
	"gopkg.in/yaml.v2"
	"k8s.io/klog"
)

var (
	customResourcePriority = &CustomAllocationPriority{"CustomResourceAllocation", customResourceScorer}

	// LeastRequestedPriorityMap is a priority function that favors nodes with fewer requested resources.
	// It calculates the percentage of memory and CPU requested by pods scheduled on the node, and
	// prioritizes based on the minimum of the average of the fraction of requested to capacity.
	//
	// Details:
	// (cpu((capacity-sum(requested))*10/capacity) + memory((capacity-sum(requested))*10/capacity))/2
	CustomRequestedPriorityMap = customResourcePriority.PriorityMap
)

type Config struct {
	Server struct {
		Port string `yaml:"port"`
		Host string `yaml:"host"`
	} `yaml:"server"`
	Database struct {
		Type     string `yaml: "type"`
		Name     string `yaml:"name"`
		Username string `yaml:"username"`
		Password string `yaml:"password"`
	} `yaml:"database"`
}

type System struct {
	ID         int    `json:"id"`
	Uuid       string `json:"uuid"`
	numSockets int    `json:"num_sockets"`
	numCores   int    `json:"num_cores`
}

var nodes = map[string]string{
	"kube-01": "c4766d29-4dc1-11ea-9d98-0242ac110002",
	"kube-02": "2",
	"kube-03": "2",
	"kube-04": "2",
	"kube-05": "2",
}

func readFile(cfg *Config, file string) {
	f, err := os.Open("/etc/kubernetes/scheduler-monitoringDB.yaml")
	if err != nil {
		panic(err.Error())
	}
	defer f.Close()

	decoder := yaml.NewDecoder(f)
	err = decoder.Decode(&cfg)
	if err != nil {
		panic(err.Error())
	}
}

func customResourceScorer(nodeName string) int64 {
	//return (customRequestedScore(requested.MilliCPU, allocable.MilliCPU) +
	//customRequestedScore(requested.Memory, allocable.Memory)) / 2

	//read database information
	var cfg Config
	readFile(&cfg, "/etc/kubernetes/scheduler-monitoringDB.yaml")
	/*-------------------------------------
	//TODO read also nodes to uuid mappings
	-------------------------------------*/

	//Access the Database
	DBstring := cfg.Database.Username + ":" + cfg.Database.Password + "@tcp(" + cfg.Server.Host + ":" + cfg.Server.Port + ")/" + cfg.Database.Name
	db, err := sql.Open(cfg.Database.Type, DBstring)
	if err != nil {
		panic(err.Error())
	}

	//TODO InfluxDB
	c, err := client.NewHttpClient(client.HTTPConfig{
		Addr: "http://" + cfg.Server.Host + ":" + cfg.Server.Port + "",
	})
	if err != nil {
		fmt.Println("Error while creating InfluxDB client: ", err.Error())
	}

	// close the connection in the end of execution
	defer c.Close()

	//Get the uuid of this node in order to query in the database
	curr_uuid, ok := nodes[nodeName]

	if ok {
		command := fmt.Sprintf("SELECT ipc from system_metrics where uuid = %s order by time desc limit 1", curr_uuid)
		q := client.NewQuery(command, "evolve", "")
		if response, err := c.Query(q); err == nil && response.Error() == nil {
			fmt.Println(response.Results)
		}
	}

	// //Close the database connection in the end of the execution
	// defer db.Close()

	// //Get the uuid of this node in order to query in the database
	// curr_uuid, ok := nodes[nodeName]

	// //Get the metrics for the current node
	// if ok {
	// 	results, err := db.Query("SELECT id, num_sockets, num_cores FROM systems WHERE uuid = ?", curr_uuid)

	// 	if err != nil {
	// 		panic(err.Error()) // proper error handling instead of panic in your app
	// 	}
	// 	sys := System{}
	// 	sys.Uuid = curr_uuid

	// 	for results.Next() {
	// 		// for each row, scan the result into our tag composite object
	// 		err = results.Scan(&sys.ID, &sys.numSockets, &sys.numCores)
	// 		if err != nil {
	// 			panic(err.Error()) // proper error handling instead of panic in your app
	// 		}
	// 		// and then print out the tag's Name attribute
	// 		klog.Infof("This is the system with name: %s, id: %d and  number of cores: %d", nodeName, sys.ID, sys.numCores)
	// 	}
	// }
	res := customRequestedScore(nodeName)
	klog.Infof("Node name %s, has score %d\n", nodeName, res)
	return res
}

func customRequestedScore(nodeName string) int64 {
	if nodeName == "kube-01" {
		return 10
	}
	return 0
}