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 (
	"encoding/json"
	"fmt"
	"os"
	"strings"

	_ "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}
	//customResourcePriority = &CustomAllocationPriority{"CustomRequestedPriority", 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"`
	MonitoringSpecs struct {
		TimeInterval float32 `yaml:"interval"`
	} `yaml:"monitoring"`
}

// type Row struct {
// 	ipc    float32
// 	l3m    float32
// 	reads  float32
// 	writes float32
// 	c6res  float32
// }

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

// TODO:
// place those maps in another file inside the package
var nodes = map[string]string{
	"kube-01": "c4766d29-4dc1-11ea-9d98-0242ac110002",
	"kube-02": "c4766d29-4dc1-11ea-9d98-0242ac110002",
	"kube-03": "c4766d29-4dc1-11ea-9d98-0242ac110002",
	"kube-04": "c4766d29-4dc1-11ea-9d98-0242ac110002",
	"kube-05": "c4766d29-4dc1-11ea-9d98-0242ac110002",
	"kube-06": "c4766d29-4dc1-11ea-9d98-0242ac110002",
	"kube-07": "c4766d29-4dc1-11ea-9d98-0242ac110002",
	"kube-08": "c4766d29-4dc1-11ea-9d98-0242ac110002",
}

var sockets = map[string]int{
	"kube-01": 0,
	"kube-02": 0,
	"kube-03": 0,
	"kube-04": 0,
	"kube-05": 0,
	"kube-06": 1,
	"kube-07": 0,
	"kube-08": 1,
}

func readFile(cfg *Config, file string) error {
	f, err := os.Open(file)
	if err != nil {
		klog.Infof("Config file for scheduler not found. Error: %v", err)
		return err
	}
	defer f.Close()

	decoder := yaml.NewDecoder(f)
	err = decoder.Decode(&cfg)
	if err != nil {
		klog.Infof("Unable to decode the config file. Error: %v", err)
		return err
	}
	return nil
}

func customScoreFn(si scorerInput) float64 {
	return si.metrics["ipc"] / si.metrics["mem_read"] * si.metrics["mem_write"]
}

func onlyIPC(metrics map[string]float64) float64 {
	return metrics["ipc"]
}

func onlyL3(metrics map[string]float64) float64 {
	return 1 / metrics["l3m"]
}

func onlyNrg(metrics map[string]float64) float64 {
	return 1 / metrics["procnrg"]
}

func calculateScore(results map[string]float64,
	logicFn func(scorerInput) float64) float64 {

	res := logicFn(scorerInput{metrics: results})
	//klog.Infof("Has score (in float) %v\n", res)

	return res
}

func calculateWeightedAverage(response *client.Response,
	numberOfRows, numberOfMetrics int) (map[string]float64, error) {
	// initialize the metrics map with a constant size
	metrics := make(map[string]float64, numberOfMetrics)
	rows := response.Results[0].Series[0]
	for i := 1; i < len(rows.Columns); i++ {
		for j := 0; j < numberOfRows; j++ {
			val, err := rows.Values[j][i].(json.Number).Float64()
			if err != nil {
				klog.Infof("Error while calculating %v", rows.Columns[i])
				return nil, err
			}
			metrics[rows.Columns[i]] += val * float64(numberOfRows-j)
		}
		metrics[rows.Columns[i]] = metrics[rows.Columns[i]] / float64((numberOfRows * (numberOfRows + 1) / 2))
		klog.Infof("%v : %v", rows.Columns[i], metrics[rows.Columns[i]])
	}
	// TODO better handling for the returning errors
	return metrics, nil
}

func connectToInfluxDB(cfg Config) (client.Client, error) {
	c, err := client.NewHTTPClient(client.HTTPConfig{
		Addr: "http://" + cfg.Server.Host + ":" + cfg.Server.Port + "",
	})
	if err != nil {
		klog.Infof("Error while connecting to InfluxDB: %v ", err.Error())
		return nil, err
	}
	klog.Infof("Connected Successfully to InfluxDB")
	return c, nil

}

func queryInfluxDB(metrics []string, uuid string, socket,
	time int, cfg Config, c client.Client) (map[string]float64, error) {

	// calculate the number of rows needed
	// i.e. 20sec / 0.5s interval => 40rows
	numberOfRows := int(float32(time) / cfg.MonitoringSpecs.TimeInterval)
	// merge all the required columns
	columns := strings.Join(metrics, ", ")
	// build the coommand
	command := fmt.Sprintf("SELECT %s from socket_metrics where uuid = '%s' and socket_id='%d' order by time desc limit %d", columns, uuid, socket, numberOfRows)
	q := client.NewQuery(command, cfg.Database.Name, "")
	response, err := c.Query(q)
	if err != nil {
		klog.Infof("Error while executing the query: %v", err.Error())
		return nil, err
	}
	// Calculate the average for the metrics provided
	return calculateWeightedAverage(response, numberOfRows, len(metrics))
}

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

	//read database information
	var cfg Config
	err := readFile(&cfg, "/etc/kubernetes/scheduler-monitoringDB.yaml")
	if err != nil {
		return 0, err
	}

	/*-------------------------------------
	//TODO read also nodes to uuid mappings for EVOLVE
	-------------------------------------*/

	// InfluxDB
	c, err := connectToInfluxDB(cfg)
	if err != nil {
		return 0, err
	}
	// 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]
	socket, _ := sockets[nodeName]

	if ok {

		// Select Socket
		results, err := queryInfluxDB([]string{"ipc", "mem_read", "mem_write"}, curr_uuid, socket, 20, cfg, c)
		if err != nil {
			klog.Infof("Error in querying or calculating average: %v", err.Error())
			return 0, nil
		}
		res := calculateScore(results, customScoreFn)

		// Select Node

		klog.Infof("Node name %s, has score %v\n", nodeName, res)
		return res, nil
	} else {
		klog.Infof("Error finding the uuid: %v", ok)
		return 0, nil
	}
}