custom_resource_allocation.go 9.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296
  1. /*
  2. Copyright 2020 Achilleas Tzenetopoulos.
  3. Licensed under the Apache License, Version 2.0 (the "License");
  4. you may not use this file except in compliance with the License.
  5. You may obtain a copy of the License at
  6. http://www.apache.org/licenses/LICENSE-2.0
  7. Unless required by applicable law or agreed to in writing, software
  8. distributed under the License is distributed on an "AS IS" BASIS,
  9. WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  10. See the License for the specific language governing permissions and
  11. limitations under the License.
  12. */
  13. package priorities
  14. import (
  15. "encoding/json"
  16. "fmt"
  17. "strings"
  18. _ "github.com/go-sql-driver/mysql"
  19. client "github.com/influxdata/influxdb1-client/v2"
  20. "github.com/iwita/kube-scheduler/customcache"
  21. "k8s.io/klog"
  22. )
  23. var (
  24. customResourcePriority = &CustomAllocationPriority{"CustomResourceAllocation", customResourceScorer}
  25. //customResourcePriority = &CustomAllocationPriority{"CustomRequestedPriority", customResourceScorer}
  26. // LeastRequestedPriorityMap is a priority function that favors nodes with fewer requested resources.
  27. // It calculates the percentage of memory and CPU requested by pods scheduled on the node, and
  28. // prioritizes based on the minimum of the average of the fraction of requested to capacity.
  29. //
  30. // Details:
  31. // (cpu((capacity-sum(requested))*10/capacity) + memory((capacity-sum(requested))*10/capacity))/2
  32. CustomRequestedPriorityMap = customResourcePriority.PriorityMap
  33. )
  34. func customScoreFn(si scorerInput) float64 {
  35. return si.metrics["ipc"] / (si.metrics["mem_read"] + si.metrics["mem_write"])
  36. }
  37. func onlyIPC(metrics map[string]float64) float64 {
  38. return metrics["ipc"]
  39. }
  40. func onlyL3(metrics map[string]float64) float64 {
  41. return 1 / metrics["l3m"]
  42. }
  43. func onlyNrg(metrics map[string]float64) float64 {
  44. return 1 / metrics["procnrg"]
  45. }
  46. func calculateScore(si scorerInput,
  47. logicFn func(scorerInput) float64) float64 {
  48. res := logicFn(si)
  49. //klog.Infof("Has score (in float) %v\n", res)
  50. return res
  51. }
  52. func calculateWeightedAverage(response *client.Response,
  53. numberOfRows, numberOfMetrics int) (map[string]float64, error) {
  54. // initialize the metrics map with a constant size
  55. metrics := make(map[string]float64, numberOfMetrics)
  56. rows := response.Results[0].Series[0]
  57. for i := 1; i < len(rows.Columns); i++ {
  58. for j := 0; j < numberOfRows; j++ {
  59. val, err := rows.Values[j][i].(json.Number).Float64()
  60. if err != nil {
  61. klog.Infof("Error while calculating %v", rows.Columns[i])
  62. return nil, err
  63. }
  64. metrics[rows.Columns[i]] += val * float64(numberOfRows-j)
  65. }
  66. metrics[rows.Columns[i]] = metrics[rows.Columns[i]] / float64((numberOfRows * (numberOfRows + 1) / 2))
  67. //klog.Infof("%v : %v", rows.Columns[i], metrics[rows.Columns[i]])
  68. }
  69. // TODO better handling for the returning errors
  70. return metrics, nil
  71. }
  72. func customScoreInfluxDB(metrics []string, uuid string, socket,
  73. numberOfRows int, cfg Config, c client.Client) (map[string]float64, error) {
  74. // calculate the number of rows needed
  75. // i.e. 20sec / 0.5s interval => 40rows
  76. //numberOfRows := int(float32(time) / cfg.MonitoringSpecs.TimeInterval)
  77. // merge all the required columns
  78. columns := strings.Join(metrics, ", ")
  79. // build the coommand
  80. var command strings.Builder
  81. fmt.Fprintf(&command, "SELECT %s from socket_metrics where uuid = '%s' and socket_id='%d' order by time desc limit %d", columns, uuid, socket, numberOfRows)
  82. q := client.NewQuery(command.String(), cfg.Database.Name, "")
  83. response, err := c.Query(q)
  84. if err != nil {
  85. klog.Infof("Error while executing the query: %v", err.Error())
  86. return nil, err
  87. }
  88. // Calculate the average for the metrics provided
  89. return calculateWeightedAverage(response, numberOfRows, len(metrics))
  90. }
  91. func InvalidateCache() {
  92. // Check if the cache needs update
  93. select {
  94. // clean the cache if 10 seconds are passed
  95. case <-customcache.LabCache.Timeout.C:
  96. klog.Infof("Time to erase")
  97. klog.Infof("Cache: %v", customcache.LabCache.Cache)
  98. //customcache.LabCache.Timeout.Stop()
  99. customcache.LabCache.CleanCache()
  100. default:
  101. }
  102. }
  103. func customResourceScorer(nodeName string) (float64, error) {
  104. //InvalidateCache()
  105. //klog.Infof("The value of the Ticker: %v", customcache.LabCache.Timeout.C)
  106. cores, _ := Cores[nodeName]
  107. var results map[string]float64
  108. // Check the cache
  109. customcache.LabCache.Mux.Lock()
  110. ipc, ok := customcache.LabCache.Cache[nodeName]["ipc"]
  111. if !ok {
  112. klog.Infof("IPC is nil")
  113. }
  114. reads, ok := customcache.LabCache.Cache[nodeName]["mem_read"]
  115. if !ok {
  116. klog.Infof("Memory Reads is nil")
  117. }
  118. writes, ok := customcache.LabCache.Cache[nodeName]["mem_write"]
  119. if !ok {
  120. klog.Infof("Memory Writes is nil")
  121. }
  122. c6res, ok := customcache.LabCache.Cache[nodeName]["c6res"]
  123. if !ok {
  124. klog.Infof("C6 state is nil")
  125. }
  126. customcache.LabCache.Mux.Unlock()
  127. // If the cache has value use it
  128. if ipc != -1 && reads != -1 && writes != -1 && c6res != -1 {
  129. results := map[string]float64{
  130. "ipc": ipc,
  131. "mem_read": reads,
  132. "mem_write": writes,
  133. "c6res": c6res,
  134. }
  135. klog.Infof("Found in the cache: ipc: %v, reads: %v, writes: %v", ipc, reads, writes)
  136. res := calculateScore(scorerInput{metrics: results}, customScoreFn)
  137. if sum := c6res * float64(len(cores)); sum < 1 {
  138. //klog.Infof("Average C6 is less than 1, so we get: %v", average["c6res"])
  139. res = res * c6res
  140. } else {
  141. res = res * 1
  142. }
  143. //Apply heterogeneity
  144. speed := links[Nodes[nodeName]][0] * links[Nodes[nodeName]][1]
  145. res = res * float64(speed)
  146. // Select Node
  147. klog.Infof("Using the cached values, Node name %s, has score %v\n", nodeName, res)
  148. return res, nil
  149. }
  150. //read database information
  151. var cfg Config
  152. err := readFile(&cfg, "/etc/kubernetes/scheduler-monitoringDB.yaml")
  153. if err != nil {
  154. return 0, err
  155. }
  156. /*-------------------------------------
  157. //TODO read also nodes to uuid mappings for EVOLVE
  158. -------------------------------------*/
  159. // InfluxDB
  160. c, err := connectToInfluxDB(cfg)
  161. if err != nil {
  162. return 0, err
  163. }
  164. // close the connection in the end of execution
  165. defer c.Close()
  166. //Get the uuid of this node in order to query in the database
  167. curr_uuid, ok := Nodes[nodeName]
  168. socket, _ := Sockets[nodeName]
  169. // cores, _ := Cores[nodeName]
  170. if ok {
  171. metrics := []string{"c6res"}
  172. time := 20
  173. numberOfRows := int(float32(time) / cfg.MonitoringSpecs.TimeInterval)
  174. // Define Core availability
  175. r, err := queryInfluxDbCores(metrics, curr_uuid, socket, numberOfRows, cfg, c, cores)
  176. if err != nil {
  177. klog.Infof("Error in querying or calculating core availability in the first stage: %v", err.Error())
  178. }
  179. average, err := calculateWeightedAverageCores(r, numberOfRows, len(metrics), len(cores))
  180. if err != nil {
  181. klog.Infof("Error defining core availability")
  182. }
  183. // Select Socket
  184. results, err = customScoreInfluxDB([]string{"ipc", "mem_read", "mem_write"}, curr_uuid, socket, numberOfRows, cfg, c)
  185. if err != nil {
  186. klog.Infof("Error in querying or calculating average for the custom score in the first stage: %v", err.Error())
  187. return 0, nil
  188. }
  189. res := calculateScore(scorerInput{metrics: results}, customScoreFn)
  190. //klog.Infof("Node: %v\t res before: %v", nodeName, res)
  191. if sum := average["c6res"] * float64(len(cores)); sum < 1 {
  192. //klog.Infof("Average C6 is less than 1, so we get: %v", average["c6res"])
  193. res = res * average["c6res"]
  194. } else {
  195. res = res * 1
  196. }
  197. //Update the cache with the new metrics
  198. err = customcache.LabCache.UpdateCache(results, average["c6res"], nodeName)
  199. if err != nil {
  200. klog.Infof(err.Error())
  201. } else {
  202. klog.Infof("Cache updated successfully for %v", nodeName)
  203. }
  204. //Apply heterogeneity
  205. speed := links[Nodes[nodeName]][0] * links[Nodes[nodeName]][1]
  206. res = res * float64(speed)
  207. // Select Node
  208. klog.Infof("Node name %s, has score %v\n", nodeName, res)
  209. return res, nil
  210. } else {
  211. klog.Infof("Error finding the uuid: %v", ok)
  212. return 0, nil
  213. }
  214. }
  215. // WARNING
  216. // c6res is not a dependable metric for isnpecting core availability
  217. // Some Systems use higher core states (e.g c7res)
  218. // func findAvailability(response *client.Response, numberOfMetrics, numberOfRows, numberOfCores int, floor float64) (map[string]float64, error) {
  219. // // initialize the metrics map with a constant size
  220. // metrics := make(map[string]float64, numberOfMetrics)
  221. // rows := response.Results[0].Series[0]
  222. // for i := 1; i < len(rows.Columns); i++ {
  223. // //klog.Infof("Name of column %v : %v\nrange of values: %v\nnumber of rows: %v\nnumber of cores %v\n", i, rows.Columns[i], len(rows.Values), numberOfRows, numberOfCores)
  224. // for j := 0; j < numberOfRows; j++ {
  225. // //avg, max := 0.0, 0.0
  226. // for k := 0; k < numberOfCores; k++ {
  227. // val, err := rows.Values[j*numberOfCores+k][i].(json.Number).Float64()
  228. // if err != nil {
  229. // klog.Infof("Error while calculating %v", rows.Columns[i])
  230. // return false, err
  231. // }
  232. // // if val > floor {
  233. // // return true, nil
  234. // // }
  235. // // sum += val
  236. // //avg += val / float64(numberOfCores)
  237. // avg += val
  238. // }
  239. // metrics[rows.Columns[i]] += avg * float64(numberOfRows-j)
  240. // }
  241. // metrics[rows.Columns[i]] = metrics[rows.Columns[i]] / float64((numberOfRows * (numberOfRows + 1) / 2))
  242. // if metrics[row.Columns[i]] > 1 {
  243. // return true, nil
  244. // }
  245. // //klog.Infof("%v : %v", rows.Columns[i], metrics[rows.Columns[i]])
  246. // }
  247. // // TODO better handling for the returning errors
  248. // return false, nil
  249. // }