cluster.go 41 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160
  1. package costmodel
  2. import (
  3. "fmt"
  4. "time"
  5. "github.com/kubecost/cost-model/pkg/util/timeutil"
  6. "github.com/kubecost/cost-model/pkg/cloud"
  7. "github.com/kubecost/cost-model/pkg/env"
  8. "github.com/kubecost/cost-model/pkg/log"
  9. "github.com/kubecost/cost-model/pkg/prom"
  10. prometheus "github.com/prometheus/client_golang/api"
  11. "k8s.io/klog"
  12. )
  13. const (
  14. queryClusterCores = `sum(
  15. avg(avg_over_time(kube_node_status_capacity_cpu_cores[%s] %s)) by (node, %s) * avg(avg_over_time(node_cpu_hourly_cost[%s] %s)) by (node, %s) * 730 +
  16. avg(avg_over_time(node_gpu_hourly_cost[%s] %s)) by (node, %s) * 730
  17. ) by (%s)`
  18. queryClusterRAM = `sum(
  19. avg(avg_over_time(kube_node_status_capacity_memory_bytes[%s] %s)) by (node, %s) / 1024 / 1024 / 1024 * avg(avg_over_time(node_ram_hourly_cost[%s] %s)) by (node, %s) * 730
  20. ) by (%s)`
  21. queryStorage = `sum(
  22. avg(avg_over_time(pv_hourly_cost[%s] %s)) by (persistentvolume, %s) * 730
  23. * avg(avg_over_time(kube_persistentvolume_capacity_bytes[%s] %s)) by (persistentvolume, %s) / 1024 / 1024 / 1024
  24. ) by (%s) %s`
  25. queryTotal = `sum(avg(node_total_hourly_cost) by (node, %s)) * 730 +
  26. sum(
  27. avg(avg_over_time(pv_hourly_cost[1h])) by (persistentvolume, %s) * 730
  28. * avg(avg_over_time(kube_persistentvolume_capacity_bytes[1h])) by (persistentvolume, %s) / 1024 / 1024 / 1024
  29. ) by (%s) %s`
  30. queryNodes = `sum(avg(node_total_hourly_cost) by (node, %s)) * 730 %s`
  31. )
  32. // Costs represents cumulative and monthly cluster costs over a given duration. Costs
  33. // are broken down by cores, memory, and storage.
  34. type ClusterCosts struct {
  35. Start *time.Time `json:"startTime"`
  36. End *time.Time `json:"endTime"`
  37. CPUCumulative float64 `json:"cpuCumulativeCost"`
  38. CPUMonthly float64 `json:"cpuMonthlyCost"`
  39. CPUBreakdown *ClusterCostsBreakdown `json:"cpuBreakdown"`
  40. GPUCumulative float64 `json:"gpuCumulativeCost"`
  41. GPUMonthly float64 `json:"gpuMonthlyCost"`
  42. RAMCumulative float64 `json:"ramCumulativeCost"`
  43. RAMMonthly float64 `json:"ramMonthlyCost"`
  44. RAMBreakdown *ClusterCostsBreakdown `json:"ramBreakdown"`
  45. StorageCumulative float64 `json:"storageCumulativeCost"`
  46. StorageMonthly float64 `json:"storageMonthlyCost"`
  47. StorageBreakdown *ClusterCostsBreakdown `json:"storageBreakdown"`
  48. TotalCumulative float64 `json:"totalCumulativeCost"`
  49. TotalMonthly float64 `json:"totalMonthlyCost"`
  50. DataMinutes float64
  51. }
  52. // ClusterCostsBreakdown provides percentage-based breakdown of a resource by
  53. // categories: user for user-space (i.e. non-system) usage, system, and idle.
  54. type ClusterCostsBreakdown struct {
  55. Idle float64 `json:"idle"`
  56. Other float64 `json:"other"`
  57. System float64 `json:"system"`
  58. User float64 `json:"user"`
  59. }
  60. // NewClusterCostsFromCumulative takes cumulative cost data over a given time range, computes
  61. // the associated monthly rate data, and returns the Costs.
  62. func NewClusterCostsFromCumulative(cpu, gpu, ram, storage float64, window, offset time.Duration, dataHours float64) (*ClusterCosts, error) {
  63. start, end := timeutil.ParseTimeRange(window, offset)
  64. // If the number of hours is not given (i.e. is zero) compute one from the window and offset
  65. if dataHours == 0 {
  66. dataHours = end.Sub(start).Hours()
  67. }
  68. // Do not allow zero-length windows to prevent divide-by-zero issues
  69. if dataHours == 0 {
  70. return nil, fmt.Errorf("illegal time range: window %s, offset %s", window, offset)
  71. }
  72. cc := &ClusterCosts{
  73. Start: &start,
  74. End: &end,
  75. CPUCumulative: cpu,
  76. GPUCumulative: gpu,
  77. RAMCumulative: ram,
  78. StorageCumulative: storage,
  79. TotalCumulative: cpu + gpu + ram + storage,
  80. CPUMonthly: cpu / dataHours * (timeutil.HoursPerMonth),
  81. GPUMonthly: gpu / dataHours * (timeutil.HoursPerMonth),
  82. RAMMonthly: ram / dataHours * (timeutil.HoursPerMonth),
  83. StorageMonthly: storage / dataHours * (timeutil.HoursPerMonth),
  84. }
  85. cc.TotalMonthly = cc.CPUMonthly + cc.GPUMonthly + cc.RAMMonthly + cc.StorageMonthly
  86. return cc, nil
  87. }
  88. type Disk struct {
  89. Cluster string
  90. Name string
  91. ProviderID string
  92. Cost float64
  93. Bytes float64
  94. Local bool
  95. Start time.Time
  96. End time.Time
  97. Minutes float64
  98. Breakdown *ClusterCostsBreakdown
  99. }
  100. func ClusterDisks(client prometheus.Client, provider cloud.Provider, duration, offset time.Duration) (map[string]*Disk, error) {
  101. durationStr := fmt.Sprintf("%dm", int64(duration.Minutes()))
  102. offsetStr := fmt.Sprintf(" offset %dm", int64(offset.Minutes()))
  103. if offset < time.Minute {
  104. offsetStr = ""
  105. }
  106. // minsPerResolution determines accuracy and resource use for the following
  107. // queries. Smaller values (higher resolution) result in better accuracy,
  108. // but more expensive queries, and vice-a-versa.
  109. minsPerResolution := 1
  110. resolution := time.Duration(minsPerResolution) * time.Minute
  111. // hourlyToCumulative is a scaling factor that, when multiplied by an hourly
  112. // value, converts it to a cumulative value; i.e.
  113. // [$/hr] * [min/res]*[hr/min] = [$/res]
  114. hourlyToCumulative := float64(minsPerResolution) * (1.0 / 60.0)
  115. // TODO niko/assets how do we not hard-code this price?
  116. costPerGBHr := 0.04 / 730.0
  117. ctx := prom.NewContext(client)
  118. queryPVCost := fmt.Sprintf(`avg(avg_over_time(pv_hourly_cost[%s]%s)) by (%s, persistentvolume,provider_id)`, durationStr, offsetStr, env.GetPromClusterLabel())
  119. queryPVSize := fmt.Sprintf(`avg(avg_over_time(kube_persistentvolume_capacity_bytes[%s]%s)) by (%s, persistentvolume)`, durationStr, offsetStr, env.GetPromClusterLabel())
  120. queryActiveMins := fmt.Sprintf(`count(pv_hourly_cost) by (%s, persistentvolume)[%s:%dm]%s`, env.GetPromClusterLabel(), durationStr, minsPerResolution, offsetStr)
  121. queryLocalStorageCost := fmt.Sprintf(`sum_over_time(sum(container_fs_limit_bytes{device!="tmpfs", id="/"}) by (instance, %s)[%s:%dm]%s) / 1024 / 1024 / 1024 * %f * %f`, env.GetPromClusterLabel(), durationStr, minsPerResolution, offsetStr, hourlyToCumulative, costPerGBHr)
  122. queryLocalStorageUsedCost := fmt.Sprintf(`sum_over_time(sum(container_fs_usage_bytes{device!="tmpfs", id="/"}) by (instance, %s)[%s:%dm]%s) / 1024 / 1024 / 1024 * %f * %f`, env.GetPromClusterLabel(), durationStr, minsPerResolution, offsetStr, hourlyToCumulative, costPerGBHr)
  123. queryLocalStorageBytes := fmt.Sprintf(`avg_over_time(sum(container_fs_limit_bytes{device!="tmpfs", id="/"}) by (instance, %s)[%s:%dm]%s)`, env.GetPromClusterLabel(), durationStr, minsPerResolution, offsetStr)
  124. queryLocalActiveMins := fmt.Sprintf(`count(node_total_hourly_cost) by (%s, node)[%s:%dm]%s`, env.GetPromClusterLabel(), durationStr, minsPerResolution, offsetStr)
  125. resChPVCost := ctx.Query(queryPVCost)
  126. resChPVSize := ctx.Query(queryPVSize)
  127. resChActiveMins := ctx.Query(queryActiveMins)
  128. resChLocalStorageCost := ctx.Query(queryLocalStorageCost)
  129. resChLocalStorageUsedCost := ctx.Query(queryLocalStorageUsedCost)
  130. resChLocalStorageBytes := ctx.Query(queryLocalStorageBytes)
  131. resChLocalActiveMins := ctx.Query(queryLocalActiveMins)
  132. resPVCost, _ := resChPVCost.Await()
  133. resPVSize, _ := resChPVSize.Await()
  134. resActiveMins, _ := resChActiveMins.Await()
  135. resLocalStorageCost, _ := resChLocalStorageCost.Await()
  136. resLocalStorageUsedCost, _ := resChLocalStorageUsedCost.Await()
  137. resLocalStorageBytes, _ := resChLocalStorageBytes.Await()
  138. resLocalActiveMins, _ := resChLocalActiveMins.Await()
  139. if ctx.HasErrors() {
  140. return nil, ctx.ErrorCollection()
  141. }
  142. diskMap := map[string]*Disk{}
  143. pvCosts(diskMap, resActiveMins, resPVSize, resPVCost)
  144. for _, result := range resLocalStorageCost {
  145. cluster, err := result.GetString(env.GetPromClusterLabel())
  146. if err != nil {
  147. cluster = env.GetClusterID()
  148. }
  149. name, err := result.GetString("instance")
  150. if err != nil {
  151. log.Warningf("ClusterDisks: local storage data missing instance")
  152. continue
  153. }
  154. cost := result.Values[0].Value
  155. key := fmt.Sprintf("%s/%s", cluster, name)
  156. if _, ok := diskMap[key]; !ok {
  157. diskMap[key] = &Disk{
  158. Cluster: cluster,
  159. Name: name,
  160. Breakdown: &ClusterCostsBreakdown{},
  161. Local: true,
  162. }
  163. }
  164. diskMap[key].Cost += cost
  165. }
  166. for _, result := range resLocalStorageUsedCost {
  167. cluster, err := result.GetString(env.GetPromClusterLabel())
  168. if err != nil {
  169. cluster = env.GetClusterID()
  170. }
  171. name, err := result.GetString("instance")
  172. if err != nil {
  173. log.Warningf("ClusterDisks: local storage usage data missing instance")
  174. continue
  175. }
  176. cost := result.Values[0].Value
  177. key := fmt.Sprintf("%s/%s", cluster, name)
  178. if _, ok := diskMap[key]; !ok {
  179. diskMap[key] = &Disk{
  180. Cluster: cluster,
  181. Name: name,
  182. Breakdown: &ClusterCostsBreakdown{},
  183. Local: true,
  184. }
  185. }
  186. diskMap[key].Breakdown.System = cost / diskMap[key].Cost
  187. }
  188. for _, result := range resLocalStorageBytes {
  189. cluster, err := result.GetString(env.GetPromClusterLabel())
  190. if err != nil {
  191. cluster = env.GetClusterID()
  192. }
  193. name, err := result.GetString("instance")
  194. if err != nil {
  195. log.Warningf("ClusterDisks: local storage data missing instance")
  196. continue
  197. }
  198. bytes := result.Values[0].Value
  199. key := fmt.Sprintf("%s/%s", cluster, name)
  200. if _, ok := diskMap[key]; !ok {
  201. diskMap[key] = &Disk{
  202. Cluster: cluster,
  203. Name: name,
  204. Breakdown: &ClusterCostsBreakdown{},
  205. Local: true,
  206. }
  207. }
  208. diskMap[key].Bytes = bytes
  209. }
  210. for _, result := range resLocalActiveMins {
  211. cluster, err := result.GetString(env.GetPromClusterLabel())
  212. if err != nil {
  213. cluster = env.GetClusterID()
  214. }
  215. name, err := result.GetString("node")
  216. if err != nil {
  217. log.Warningf("ClusterDisks: local active mins data missing instance")
  218. continue
  219. }
  220. key := fmt.Sprintf("%s/%s", cluster, name)
  221. if _, ok := diskMap[key]; !ok {
  222. log.Warningf("ClusterDisks: local active mins for unidentified disk")
  223. continue
  224. }
  225. if len(result.Values) == 0 {
  226. continue
  227. }
  228. s := time.Unix(int64(result.Values[0].Timestamp), 0)
  229. e := time.Unix(int64(result.Values[len(result.Values)-1].Timestamp), 0).Add(resolution)
  230. mins := e.Sub(s).Minutes()
  231. // TODO niko/assets if mins >= threshold, interpolate for missing data?
  232. diskMap[key].End = e
  233. diskMap[key].Start = s
  234. diskMap[key].Minutes = mins
  235. }
  236. for _, disk := range diskMap {
  237. // Apply all remaining RAM to Idle
  238. disk.Breakdown.Idle = 1.0 - (disk.Breakdown.System + disk.Breakdown.Other + disk.Breakdown.User)
  239. // Set provider Id to the name for reconciliation on Azure
  240. if fmt.Sprintf("%T", provider) == "*provider.Azure" {
  241. if disk.ProviderID == "" {
  242. disk.ProviderID = disk.Name
  243. }
  244. }
  245. }
  246. return diskMap, nil
  247. }
  248. func pvCosts(diskMap map[string]*Disk, resActiveMins, resPVSize, resPVCost []*prom.QueryResult) {
  249. for _, result := range resActiveMins {
  250. cluster, err := result.GetString(env.GetPromClusterLabel())
  251. if err != nil {
  252. cluster = env.GetClusterID()
  253. }
  254. name, err := result.GetString("persistentvolume")
  255. if err != nil {
  256. log.Warningf("ClusterDisks: active mins missing pv name")
  257. continue
  258. }
  259. if len(result.Values) == 0 {
  260. continue
  261. }
  262. key := fmt.Sprintf("%s/%s", cluster, name)
  263. if _, ok := diskMap[key]; !ok {
  264. diskMap[key] = &Disk{
  265. Cluster: cluster,
  266. Name: name,
  267. Breakdown: &ClusterCostsBreakdown{},
  268. }
  269. }
  270. s := time.Unix(int64(result.Values[0].Timestamp), 0)
  271. e := time.Unix(int64(result.Values[len(result.Values)-1].Timestamp), 0).Add(resolution)
  272. mins := e.Sub(s).Minutes()
  273. // TODO niko/assets if mins >= threshold, interpolate for missing data?
  274. diskMap[key].End = e
  275. diskMap[key].Start = s
  276. diskMap[key].Minutes = mins
  277. }
  278. for _, result := range resPVSize {
  279. cluster, err := result.GetString(env.GetPromClusterLabel())
  280. if err != nil {
  281. cluster = env.GetClusterID()
  282. }
  283. name, err := result.GetString("persistentvolume")
  284. if err != nil {
  285. log.Warningf("ClusterDisks: PV size data missing persistentvolume")
  286. continue
  287. }
  288. // TODO niko/assets storage class
  289. bytes := result.Values[0].Value
  290. key := fmt.Sprintf("%s/%s", cluster, name)
  291. if _, ok := diskMap[key]; !ok {
  292. diskMap[key] = &Disk{
  293. Cluster: cluster,
  294. Name: name,
  295. Breakdown: &ClusterCostsBreakdown{},
  296. }
  297. }
  298. diskMap[key].Bytes = bytes
  299. }
  300. for _, result := range resPVCost {
  301. cluster, err := result.GetString(env.GetPromClusterLabel())
  302. if err != nil {
  303. cluster = env.GetClusterID()
  304. }
  305. name, err := result.GetString("persistentvolume")
  306. if err != nil {
  307. log.Warningf("ClusterDisks: PV cost data missing persistentvolume")
  308. continue
  309. }
  310. // TODO niko/assets storage class
  311. cost := result.Values[0].Value
  312. key := fmt.Sprintf("%s/%s", cluster, name)
  313. if _, ok := diskMap[key]; !ok {
  314. diskMap[key] = &Disk{
  315. Cluster: cluster,
  316. Name: name,
  317. Breakdown: &ClusterCostsBreakdown{},
  318. }
  319. }
  320. diskMap[key].Cost = cost * (diskMap[key].Bytes / 1024 / 1024 / 1024) * (diskMap[key].Minutes / 60)
  321. providerID, _ := result.GetString("provider_id") // just put the providerID set up here, it's the simplest query.
  322. if providerID != "" {
  323. diskMap[key].ProviderID = cloud.ParsePVID(providerID)
  324. }
  325. }
  326. }
  327. type Node struct {
  328. Cluster string
  329. Name string
  330. ProviderID string
  331. NodeType string
  332. CPUCost float64
  333. CPUCores float64
  334. GPUCost float64
  335. GPUCount float64
  336. RAMCost float64
  337. RAMBytes float64
  338. Discount float64
  339. Preemptible bool
  340. CPUBreakdown *ClusterCostsBreakdown
  341. RAMBreakdown *ClusterCostsBreakdown
  342. Start time.Time
  343. End time.Time
  344. Minutes float64
  345. Labels map[string]string
  346. CostPerCPUHr float64
  347. CostPerRAMGiBHr float64
  348. CostPerGPUHr float64
  349. }
  350. // GKE lies about the number of cores e2 nodes have. This table
  351. // contains a mapping from node type -> actual CPU cores
  352. // for those cases.
  353. var partialCPUMap = map[string]float64{
  354. "e2-micro": 0.25,
  355. "e2-small": 0.5,
  356. "e2-medium": 1.0,
  357. }
  358. type NodeIdentifier struct {
  359. Cluster string
  360. Name string
  361. ProviderID string
  362. }
  363. type nodeIdentifierNoProviderID struct {
  364. Cluster string
  365. Name string
  366. }
  367. func costTimesMinuteAndCount(activeDataMap map[NodeIdentifier]activeData, costMap map[NodeIdentifier]float64, resourceCountMap map[nodeIdentifierNoProviderID]float64) {
  368. for k, v := range activeDataMap {
  369. keyNon := nodeIdentifierNoProviderID{
  370. Cluster: k.Cluster,
  371. Name: k.Name,
  372. }
  373. if cost, ok := costMap[k]; ok {
  374. minutes := v.minutes
  375. count := 1.0
  376. if c, ok := resourceCountMap[keyNon]; ok {
  377. count = c
  378. }
  379. costMap[k] = cost * (minutes / 60) * count
  380. }
  381. }
  382. }
  383. func costTimesMinute(activeDataMap map[NodeIdentifier]activeData, costMap map[NodeIdentifier]float64) {
  384. for k, v := range activeDataMap {
  385. if cost, ok := costMap[k]; ok {
  386. minutes := v.minutes
  387. costMap[k] = cost * (minutes / 60)
  388. }
  389. }
  390. }
  391. func ClusterNodes(cp cloud.Provider, client prometheus.Client, duration, offset time.Duration) (map[NodeIdentifier]*Node, error) {
  392. durationStr := fmt.Sprintf("%dm", int64(duration.Minutes()))
  393. offsetStr := fmt.Sprintf(" offset %dm", int64(offset.Minutes()))
  394. if offset < time.Minute {
  395. offsetStr = ""
  396. }
  397. // minsPerResolution determines accuracy and resource use for the following
  398. // queries. Smaller values (higher resolution) result in better accuracy,
  399. // but more expensive queries, and vice-a-versa.
  400. minsPerResolution := 1
  401. resolution := time.Duration(minsPerResolution) * time.Minute
  402. requiredCtx := prom.NewContext(client)
  403. optionalCtx := prom.NewContext(client)
  404. queryNodeCPUHourlyCost := fmt.Sprintf(`avg(avg_over_time(node_cpu_hourly_cost[%s]%s)) by (%s, node, instance_type, provider_id)`, durationStr, offsetStr, env.GetPromClusterLabel())
  405. queryNodeCPUCores := fmt.Sprintf(`avg(avg_over_time(kube_node_status_capacity_cpu_cores[%s]%s)) by (%s, node)`, durationStr, offsetStr, env.GetPromClusterLabel())
  406. queryNodeRAMHourlyCost := fmt.Sprintf(`avg(avg_over_time(node_ram_hourly_cost[%s]%s)) by (%s, node, instance_type, provider_id) / 1024 / 1024 / 1024`, durationStr, offsetStr, env.GetPromClusterLabel())
  407. queryNodeRAMBytes := fmt.Sprintf(`avg(avg_over_time(kube_node_status_capacity_memory_bytes[%s]%s)) by (%s, node)`, durationStr, offsetStr, env.GetPromClusterLabel())
  408. queryNodeGPUCount := fmt.Sprintf(`avg(avg_over_time(node_gpu_count[%s]%s)) by (%s, node, provider_id)`, durationStr, offsetStr, env.GetPromClusterLabel())
  409. queryNodeGPUHourlyCost := fmt.Sprintf(`avg(avg_over_time(node_gpu_hourly_cost[%s]%s)) by (%s, node, instance_type, provider_id)`, durationStr, offsetStr, env.GetPromClusterLabel())
  410. queryNodeCPUModeTotal := fmt.Sprintf(`sum(rate(node_cpu_seconds_total[%s:%dm]%s)) by (kubernetes_node, %s, mode)`, durationStr, minsPerResolution, offsetStr, env.GetPromClusterLabel())
  411. queryNodeRAMSystemPct := fmt.Sprintf(`sum(sum_over_time(container_memory_working_set_bytes{container_name!="POD",container_name!="",namespace="kube-system"}[%s:%dm]%s)) by (instance, %s) / avg(label_replace(sum(sum_over_time(kube_node_status_capacity_memory_bytes[%s:%dm]%s)) by (node, %s), "instance", "$1", "node", "(.*)")) by (instance, %s)`, durationStr, minsPerResolution, offsetStr, env.GetPromClusterLabel(), durationStr, minsPerResolution, offsetStr, env.GetPromClusterLabel(), env.GetPromClusterLabel())
  412. queryNodeRAMUserPct := fmt.Sprintf(`sum(sum_over_time(container_memory_working_set_bytes{container_name!="POD",container_name!="",namespace!="kube-system"}[%s:%dm]%s)) by (instance, %s) / avg(label_replace(sum(sum_over_time(kube_node_status_capacity_memory_bytes[%s:%dm]%s)) by (node, %s), "instance", "$1", "node", "(.*)")) by (instance, %s)`, durationStr, minsPerResolution, offsetStr, env.GetPromClusterLabel(), durationStr, minsPerResolution, offsetStr, env.GetPromClusterLabel(), env.GetPromClusterLabel())
  413. queryActiveMins := fmt.Sprintf(`avg(node_total_hourly_cost) by (node, %s, provider_id)[%s:%dm]%s`, env.GetPromClusterLabel(), durationStr, minsPerResolution, offsetStr)
  414. queryIsSpot := fmt.Sprintf(`avg_over_time(kubecost_node_is_spot[%s:%dm]%s)`, durationStr, minsPerResolution, offsetStr)
  415. queryLabels := fmt.Sprintf(`count_over_time(kube_node_labels[%s:%dm]%s)`, durationStr, minsPerResolution, offsetStr)
  416. // Return errors if these fail
  417. resChNodeCPUHourlyCost := requiredCtx.Query(queryNodeCPUHourlyCost)
  418. resChNodeCPUCores := requiredCtx.Query(queryNodeCPUCores)
  419. resChNodeRAMHourlyCost := requiredCtx.Query(queryNodeRAMHourlyCost)
  420. resChNodeRAMBytes := requiredCtx.Query(queryNodeRAMBytes)
  421. resChNodeGPUCount := requiredCtx.Query(queryNodeGPUCount)
  422. resChNodeGPUHourlyCost := requiredCtx.Query(queryNodeGPUHourlyCost)
  423. resChActiveMins := requiredCtx.Query(queryActiveMins)
  424. resChIsSpot := requiredCtx.Query(queryIsSpot)
  425. // Do not return errors if these fail, but log warnings
  426. resChNodeCPUModeTotal := optionalCtx.Query(queryNodeCPUModeTotal)
  427. resChNodeRAMSystemPct := optionalCtx.Query(queryNodeRAMSystemPct)
  428. resChNodeRAMUserPct := optionalCtx.Query(queryNodeRAMUserPct)
  429. resChLabels := optionalCtx.Query(queryLabels)
  430. resNodeCPUHourlyCost, _ := resChNodeCPUHourlyCost.Await()
  431. resNodeCPUCores, _ := resChNodeCPUCores.Await()
  432. resNodeGPUCount, _ := resChNodeGPUCount.Await()
  433. resNodeGPUHourlyCost, _ := resChNodeGPUHourlyCost.Await()
  434. resNodeRAMHourlyCost, _ := resChNodeRAMHourlyCost.Await()
  435. resNodeRAMBytes, _ := resChNodeRAMBytes.Await()
  436. resIsSpot, _ := resChIsSpot.Await()
  437. resNodeCPUModeTotal, _ := resChNodeCPUModeTotal.Await()
  438. resNodeRAMSystemPct, _ := resChNodeRAMSystemPct.Await()
  439. resNodeRAMUserPct, _ := resChNodeRAMUserPct.Await()
  440. resActiveMins, _ := resChActiveMins.Await()
  441. resLabels, _ := resChLabels.Await()
  442. if optionalCtx.HasErrors() {
  443. for _, err := range optionalCtx.Errors() {
  444. log.Warningf("ClusterNodes: %s", err)
  445. }
  446. }
  447. if requiredCtx.HasErrors() {
  448. for _, err := range requiredCtx.Errors() {
  449. log.Errorf("ClusterNodes: %s", err)
  450. }
  451. return nil, requiredCtx.ErrorCollection()
  452. }
  453. activeDataMap := buildActiveDataMap(resActiveMins, resolution)
  454. gpuCountMap := buildGPUCountMap(resNodeGPUCount)
  455. cpuCostMap, clusterAndNameToType1 := buildCPUCostMap(resNodeCPUHourlyCost)
  456. ramCostMap, clusterAndNameToType2 := buildRAMCostMap(resNodeRAMHourlyCost)
  457. gpuCostMap, clusterAndNameToType3 := buildGPUCostMap(resNodeGPUHourlyCost, gpuCountMap)
  458. clusterAndNameToTypeIntermediate := mergeTypeMaps(clusterAndNameToType1, clusterAndNameToType2)
  459. clusterAndNameToType := mergeTypeMaps(clusterAndNameToTypeIntermediate, clusterAndNameToType3)
  460. cpuCoresMap := buildCPUCoresMap(resNodeCPUCores)
  461. ramBytesMap := buildRAMBytesMap(resNodeRAMBytes)
  462. ramUserPctMap := buildRAMUserPctMap(resNodeRAMUserPct)
  463. ramSystemPctMap := buildRAMSystemPctMap(resNodeRAMSystemPct)
  464. cpuBreakdownMap := buildCPUBreakdownMap(resNodeCPUModeTotal)
  465. preemptibleMap := buildPreemptibleMap(resIsSpot)
  466. labelsMap := buildLabelsMap(resLabels)
  467. costTimesMinuteAndCount(activeDataMap, cpuCostMap, cpuCoresMap)
  468. costTimesMinuteAndCount(activeDataMap, ramCostMap, ramBytesMap)
  469. costTimesMinute(activeDataMap, gpuCostMap) // there's no need to do a weird "nodeIdentifierNoProviderID" type match since gpuCounts have a providerID
  470. nodeMap := buildNodeMap(
  471. cpuCostMap, ramCostMap, gpuCostMap, gpuCountMap,
  472. cpuCoresMap, ramBytesMap, ramUserPctMap,
  473. ramSystemPctMap,
  474. cpuBreakdownMap,
  475. activeDataMap,
  476. preemptibleMap,
  477. labelsMap,
  478. clusterAndNameToType,
  479. )
  480. c, err := cp.GetConfig()
  481. if err != nil {
  482. return nil, err
  483. }
  484. discount, err := ParsePercentString(c.Discount)
  485. if err != nil {
  486. return nil, err
  487. }
  488. negotiatedDiscount, err := ParsePercentString(c.NegotiatedDiscount)
  489. if err != nil {
  490. return nil, err
  491. }
  492. for _, node := range nodeMap {
  493. // TODO take GKE Reserved Instances into account
  494. node.Discount = cp.CombinedDiscountForNode(node.NodeType, node.Preemptible, discount, negotiatedDiscount)
  495. // Apply all remaining resources to Idle
  496. node.CPUBreakdown.Idle = 1.0 - (node.CPUBreakdown.System + node.CPUBreakdown.Other + node.CPUBreakdown.User)
  497. node.RAMBreakdown.Idle = 1.0 - (node.RAMBreakdown.System + node.RAMBreakdown.Other + node.RAMBreakdown.User)
  498. }
  499. return nodeMap, nil
  500. }
  501. type LoadBalancer struct {
  502. Cluster string
  503. Name string
  504. ProviderID string
  505. Cost float64
  506. Start time.Time
  507. Minutes float64
  508. }
  509. func ClusterLoadBalancers(client prometheus.Client, duration, offset time.Duration) (map[string]*LoadBalancer, error) {
  510. durationStr := fmt.Sprintf("%dm", int64(duration.Minutes()))
  511. offsetStr := fmt.Sprintf(" offset %dm", int64(offset.Minutes()))
  512. if offset < time.Minute {
  513. offsetStr = ""
  514. }
  515. // minsPerResolution determines accuracy and resource use for the following
  516. // queries. Smaller values (higher resolution) result in better accuracy,
  517. // but more expensive queries, and vice-a-versa.
  518. minsPerResolution := 5
  519. // hourlyToCumulative is a scaling factor that, when multiplied by an hourly
  520. // value, converts it to a cumulative value; i.e.
  521. // [$/hr] * [min/res]*[hr/min] = [$/res]
  522. hourlyToCumulative := float64(minsPerResolution) * (1.0 / 60.0)
  523. ctx := prom.NewContext(client)
  524. queryLBCost := fmt.Sprintf(`sum_over_time((avg(kubecost_load_balancer_cost) by (namespace, service_name, %s, ingress_ip))[%s:%dm]%s) * %f`, env.GetPromClusterLabel(), durationStr, minsPerResolution, offsetStr, hourlyToCumulative)
  525. queryActiveMins := fmt.Sprintf(`count(kubecost_load_balancer_cost) by (namespace, service_name, %s, ingress_ip)[%s:%dm]%s`, env.GetPromClusterLabel(), durationStr, minsPerResolution, offsetStr)
  526. resChLBCost := ctx.Query(queryLBCost)
  527. resChActiveMins := ctx.Query(queryActiveMins)
  528. resLBCost, _ := resChLBCost.Await()
  529. resActiveMins, _ := resChActiveMins.Await()
  530. if ctx.HasErrors() {
  531. return nil, ctx.ErrorCollection()
  532. }
  533. loadBalancerMap := map[string]*LoadBalancer{}
  534. for _, result := range resLBCost {
  535. cluster, err := result.GetString(env.GetPromClusterLabel())
  536. if err != nil {
  537. cluster = env.GetClusterID()
  538. }
  539. namespace, err := result.GetString("namespace")
  540. if err != nil {
  541. log.Warningf("ClusterLoadBalancers: LB cost data missing namespace")
  542. continue
  543. }
  544. serviceName, err := result.GetString("service_name")
  545. if err != nil {
  546. log.Warningf("ClusterLoadBalancers: LB cost data missing service_name")
  547. continue
  548. }
  549. providerID, err := result.GetString("ingress_ip")
  550. if err != nil {
  551. log.DedupedWarningf(5, "ClusterLoadBalancers: LB cost data missing ingress_ip")
  552. providerID = ""
  553. }
  554. lbCost := result.Values[0].Value
  555. key := fmt.Sprintf("%s/%s/%s", cluster, namespace, serviceName)
  556. if _, ok := loadBalancerMap[key]; !ok {
  557. loadBalancerMap[key] = &LoadBalancer{
  558. Cluster: cluster,
  559. Name: namespace + "/" + serviceName,
  560. ProviderID: cloud.ParseLBID(providerID),
  561. }
  562. }
  563. // Fill in Provider ID if it is available and missing in the loadBalancerMap
  564. // Prevents there from being a duplicate LoadBalancers on the same day
  565. if providerID != "" && loadBalancerMap[key].ProviderID == "" {
  566. loadBalancerMap[key].ProviderID = providerID
  567. }
  568. loadBalancerMap[key].Cost += lbCost
  569. }
  570. for _, result := range resActiveMins {
  571. cluster, err := result.GetString(env.GetPromClusterLabel())
  572. if err != nil {
  573. cluster = env.GetClusterID()
  574. }
  575. namespace, err := result.GetString("namespace")
  576. if err != nil {
  577. log.Warningf("ClusterLoadBalancers: LB cost data missing namespace")
  578. continue
  579. }
  580. serviceName, err := result.GetString("service_name")
  581. if err != nil {
  582. log.Warningf("ClusterLoadBalancers: LB cost data missing service_name")
  583. continue
  584. }
  585. key := fmt.Sprintf("%s/%s/%s", cluster, namespace, serviceName)
  586. if len(result.Values) == 0 {
  587. continue
  588. }
  589. if lb, ok := loadBalancerMap[key]; ok {
  590. s := time.Unix(int64(result.Values[0].Timestamp), 0)
  591. e := time.Unix(int64(result.Values[len(result.Values)-1].Timestamp), 0)
  592. mins := e.Sub(s).Minutes()
  593. lb.Start = s
  594. lb.Minutes = mins
  595. } else {
  596. log.DedupedWarningf(20, "ClusterLoadBalancers: found minutes for key that does not exist: %s", key)
  597. }
  598. }
  599. return loadBalancerMap, nil
  600. }
  601. // ComputeClusterCosts gives the cumulative and monthly-rate cluster costs over a window of time for all clusters.
  602. func (a *Accesses) ComputeClusterCosts(client prometheus.Client, provider cloud.Provider, window, offset time.Duration, withBreakdown bool) (map[string]*ClusterCosts, error) {
  603. // Compute number of minutes in the full interval, for use interpolating missed scrapes or scaling missing data
  604. start, end := timeutil.ParseTimeRange(window, offset)
  605. mins := end.Sub(start).Minutes()
  606. // minsPerResolution determines accuracy and resource use for the following
  607. // queries. Smaller values (higher resolution) result in better accuracy,
  608. // but more expensive queries, and vice-a-versa.
  609. minsPerResolution := 5
  610. // hourlyToCumulative is a scaling factor that, when multiplied by an hourly
  611. // value, converts it to a cumulative value; i.e.
  612. // [$/hr] * [min/res]*[hr/min] = [$/res]
  613. hourlyToCumulative := float64(minsPerResolution) * (1.0 / 60.0)
  614. const fmtQueryDataCount = `
  615. count_over_time(sum(kube_node_status_capacity_cpu_cores) by (%s)[%s:%dm]%s) * %d
  616. `
  617. const fmtQueryTotalGPU = `
  618. sum(
  619. sum_over_time(node_gpu_hourly_cost[%s:%dm]%s) * %f
  620. ) by (%s)
  621. `
  622. const fmtQueryTotalCPU = `
  623. sum(
  624. sum_over_time(avg(kube_node_status_capacity_cpu_cores) by (node, %s)[%s:%dm]%s) *
  625. avg(avg_over_time(node_cpu_hourly_cost[%s:%dm]%s)) by (node, %s) * %f
  626. ) by (%s)
  627. `
  628. const fmtQueryTotalRAM = `
  629. sum(
  630. sum_over_time(avg(kube_node_status_capacity_memory_bytes) by (node, %s)[%s:%dm]%s) / 1024 / 1024 / 1024 *
  631. avg(avg_over_time(node_ram_hourly_cost[%s:%dm]%s)) by (node, %s) * %f
  632. ) by (%s)
  633. `
  634. const fmtQueryTotalStorage = `
  635. sum(
  636. sum_over_time(avg(kube_persistentvolume_capacity_bytes) by (persistentvolume, %s)[%s:%dm]%s) / 1024 / 1024 / 1024 *
  637. avg(avg_over_time(pv_hourly_cost[%s:%dm]%s)) by (persistentvolume, %s) * %f
  638. ) by (%s)
  639. `
  640. const fmtQueryCPUModePct = `
  641. sum(rate(node_cpu_seconds_total[%s]%s)) by (%s, mode) / ignoring(mode)
  642. group_left sum(rate(node_cpu_seconds_total[%s]%s)) by (%s)
  643. `
  644. const fmtQueryRAMSystemPct = `
  645. sum(sum_over_time(container_memory_usage_bytes{container_name!="",namespace="kube-system"}[%s:%dm]%s)) by (%s)
  646. / sum(sum_over_time(kube_node_status_capacity_memory_bytes[%s:%dm]%s)) by (%s)
  647. `
  648. const fmtQueryRAMUserPct = `
  649. sum(sum_over_time(kubecost_cluster_memory_working_set_bytes[%s:%dm]%s)) by (%s)
  650. / sum(sum_over_time(kube_node_status_capacity_memory_bytes[%s:%dm]%s)) by (%s)
  651. `
  652. // TODO niko/clustercost metric "kubelet_volume_stats_used_bytes" was deprecated in 1.12, then seems to have come back in 1.17
  653. // const fmtQueryPVStorageUsePct = `(sum(kube_persistentvolumeclaim_info) by (persistentvolumeclaim, storageclass,namespace) + on (persistentvolumeclaim,namespace)
  654. // group_right(storageclass) sum(kubelet_volume_stats_used_bytes) by (persistentvolumeclaim,namespace))`
  655. queryUsedLocalStorage := provider.GetLocalStorageQuery(window, offset, false, true)
  656. queryTotalLocalStorage := provider.GetLocalStorageQuery(window, offset, false, false)
  657. if queryTotalLocalStorage != "" {
  658. queryTotalLocalStorage = fmt.Sprintf(" + %s", queryTotalLocalStorage)
  659. }
  660. fmtOffset := timeutil.DurationToPromOffsetString(offset)
  661. queryDataCount := fmt.Sprintf(fmtQueryDataCount, env.GetPromClusterLabel(), window, minsPerResolution, fmtOffset, minsPerResolution)
  662. queryTotalGPU := fmt.Sprintf(fmtQueryTotalGPU, window, minsPerResolution, fmtOffset, hourlyToCumulative, env.GetPromClusterLabel())
  663. queryTotalCPU := fmt.Sprintf(fmtQueryTotalCPU, env.GetPromClusterLabel(), window, minsPerResolution, fmtOffset, window, minsPerResolution, fmtOffset, env.GetPromClusterLabel(), hourlyToCumulative, env.GetPromClusterLabel())
  664. queryTotalRAM := fmt.Sprintf(fmtQueryTotalRAM, env.GetPromClusterLabel(), window, minsPerResolution, fmtOffset, window, minsPerResolution, fmtOffset, env.GetPromClusterLabel(), hourlyToCumulative, env.GetPromClusterLabel())
  665. queryTotalStorage := fmt.Sprintf(fmtQueryTotalStorage, env.GetPromClusterLabel(), window, minsPerResolution, fmtOffset, window, minsPerResolution, fmtOffset, env.GetPromClusterLabel(), hourlyToCumulative, env.GetPromClusterLabel())
  666. ctx := prom.NewContext(client)
  667. resChs := ctx.QueryAll(
  668. queryDataCount,
  669. queryTotalGPU,
  670. queryTotalCPU,
  671. queryTotalRAM,
  672. queryTotalStorage,
  673. )
  674. // Only submit the local storage query if it is valid. Otherwise Prometheus
  675. // will return errors. Always append something to resChs, regardless, to
  676. // maintain indexing.
  677. if queryTotalLocalStorage != "" {
  678. resChs = append(resChs, ctx.Query(queryTotalLocalStorage))
  679. } else {
  680. resChs = append(resChs, nil)
  681. }
  682. if withBreakdown {
  683. queryCPUModePct := fmt.Sprintf(fmtQueryCPUModePct, window, fmtOffset, env.GetPromClusterLabel(), window, fmtOffset, env.GetPromClusterLabel())
  684. queryRAMSystemPct := fmt.Sprintf(fmtQueryRAMSystemPct, window, minsPerResolution, fmtOffset, env.GetPromClusterLabel(), window, minsPerResolution, fmtOffset, env.GetPromClusterLabel())
  685. queryRAMUserPct := fmt.Sprintf(fmtQueryRAMUserPct, window, minsPerResolution, fmtOffset, env.GetPromClusterLabel(), window, minsPerResolution, fmtOffset, env.GetPromClusterLabel())
  686. bdResChs := ctx.QueryAll(
  687. queryCPUModePct,
  688. queryRAMSystemPct,
  689. queryRAMUserPct,
  690. )
  691. // Only submit the local storage query if it is valid. Otherwise Prometheus
  692. // will return errors. Always append something to resChs, regardless, to
  693. // maintain indexing.
  694. if queryUsedLocalStorage != "" {
  695. bdResChs = append(bdResChs, ctx.Query(queryUsedLocalStorage))
  696. } else {
  697. bdResChs = append(bdResChs, nil)
  698. }
  699. resChs = append(resChs, bdResChs...)
  700. }
  701. resDataCount, _ := resChs[0].Await()
  702. resTotalGPU, _ := resChs[1].Await()
  703. resTotalCPU, _ := resChs[2].Await()
  704. resTotalRAM, _ := resChs[3].Await()
  705. resTotalStorage, _ := resChs[4].Await()
  706. if ctx.HasErrors() {
  707. return nil, ctx.ErrorCollection()
  708. }
  709. defaultClusterID := env.GetClusterID()
  710. dataMinsByCluster := map[string]float64{}
  711. for _, result := range resDataCount {
  712. clusterID, _ := result.GetString(env.GetPromClusterLabel())
  713. if clusterID == "" {
  714. clusterID = defaultClusterID
  715. }
  716. dataMins := mins
  717. if len(result.Values) > 0 {
  718. dataMins = result.Values[0].Value
  719. } else {
  720. klog.V(3).Infof("[Warning] cluster cost data count returned no results for cluster %s", clusterID)
  721. }
  722. dataMinsByCluster[clusterID] = dataMins
  723. }
  724. // Determine combined discount
  725. discount, customDiscount := 0.0, 0.0
  726. c, err := a.CloudProvider.GetConfig()
  727. if err == nil {
  728. discount, err = ParsePercentString(c.Discount)
  729. if err != nil {
  730. discount = 0.0
  731. }
  732. customDiscount, err = ParsePercentString(c.NegotiatedDiscount)
  733. if err != nil {
  734. customDiscount = 0.0
  735. }
  736. }
  737. // Intermediate structure storing mapping of [clusterID][type ∈ {cpu, ram, storage, total}]=cost
  738. costData := make(map[string]map[string]float64)
  739. // Helper function to iterate over Prom query results, parsing the raw values into
  740. // the intermediate costData structure.
  741. setCostsFromResults := func(costData map[string]map[string]float64, results []*prom.QueryResult, name string, discount float64, customDiscount float64) {
  742. for _, result := range results {
  743. clusterID, _ := result.GetString(env.GetPromClusterLabel())
  744. if clusterID == "" {
  745. clusterID = defaultClusterID
  746. }
  747. if _, ok := costData[clusterID]; !ok {
  748. costData[clusterID] = map[string]float64{}
  749. }
  750. if len(result.Values) > 0 {
  751. costData[clusterID][name] += result.Values[0].Value * (1.0 - discount) * (1.0 - customDiscount)
  752. costData[clusterID]["total"] += result.Values[0].Value * (1.0 - discount) * (1.0 - customDiscount)
  753. }
  754. }
  755. }
  756. // Apply both sustained use and custom discounts to RAM and CPU
  757. setCostsFromResults(costData, resTotalCPU, "cpu", discount, customDiscount)
  758. setCostsFromResults(costData, resTotalRAM, "ram", discount, customDiscount)
  759. // Apply only custom discount to GPU and storage
  760. setCostsFromResults(costData, resTotalGPU, "gpu", 0.0, customDiscount)
  761. setCostsFromResults(costData, resTotalStorage, "storage", 0.0, customDiscount)
  762. if queryTotalLocalStorage != "" {
  763. resTotalLocalStorage, err := resChs[5].Await()
  764. if err != nil {
  765. return nil, err
  766. }
  767. setCostsFromResults(costData, resTotalLocalStorage, "localstorage", 0.0, customDiscount)
  768. }
  769. cpuBreakdownMap := map[string]*ClusterCostsBreakdown{}
  770. ramBreakdownMap := map[string]*ClusterCostsBreakdown{}
  771. pvUsedCostMap := map[string]float64{}
  772. if withBreakdown {
  773. resCPUModePct, _ := resChs[6].Await()
  774. resRAMSystemPct, _ := resChs[7].Await()
  775. resRAMUserPct, _ := resChs[8].Await()
  776. if ctx.HasErrors() {
  777. return nil, ctx.ErrorCollection()
  778. }
  779. for _, result := range resCPUModePct {
  780. clusterID, _ := result.GetString(env.GetPromClusterLabel())
  781. if clusterID == "" {
  782. clusterID = defaultClusterID
  783. }
  784. if _, ok := cpuBreakdownMap[clusterID]; !ok {
  785. cpuBreakdownMap[clusterID] = &ClusterCostsBreakdown{}
  786. }
  787. cpuBD := cpuBreakdownMap[clusterID]
  788. mode, err := result.GetString("mode")
  789. if err != nil {
  790. klog.V(3).Infof("[Warning] ComputeClusterCosts: unable to read CPU mode: %s", err)
  791. mode = "other"
  792. }
  793. switch mode {
  794. case "idle":
  795. cpuBD.Idle += result.Values[0].Value
  796. case "system":
  797. cpuBD.System += result.Values[0].Value
  798. case "user":
  799. cpuBD.User += result.Values[0].Value
  800. default:
  801. cpuBD.Other += result.Values[0].Value
  802. }
  803. }
  804. for _, result := range resRAMSystemPct {
  805. clusterID, _ := result.GetString(env.GetPromClusterLabel())
  806. if clusterID == "" {
  807. clusterID = defaultClusterID
  808. }
  809. if _, ok := ramBreakdownMap[clusterID]; !ok {
  810. ramBreakdownMap[clusterID] = &ClusterCostsBreakdown{}
  811. }
  812. ramBD := ramBreakdownMap[clusterID]
  813. ramBD.System += result.Values[0].Value
  814. }
  815. for _, result := range resRAMUserPct {
  816. clusterID, _ := result.GetString(env.GetPromClusterLabel())
  817. if clusterID == "" {
  818. clusterID = defaultClusterID
  819. }
  820. if _, ok := ramBreakdownMap[clusterID]; !ok {
  821. ramBreakdownMap[clusterID] = &ClusterCostsBreakdown{}
  822. }
  823. ramBD := ramBreakdownMap[clusterID]
  824. ramBD.User += result.Values[0].Value
  825. }
  826. for _, ramBD := range ramBreakdownMap {
  827. remaining := 1.0
  828. remaining -= ramBD.Other
  829. remaining -= ramBD.System
  830. remaining -= ramBD.User
  831. ramBD.Idle = remaining
  832. }
  833. if queryUsedLocalStorage != "" {
  834. resUsedLocalStorage, err := resChs[9].Await()
  835. if err != nil {
  836. return nil, err
  837. }
  838. for _, result := range resUsedLocalStorage {
  839. clusterID, _ := result.GetString(env.GetPromClusterLabel())
  840. if clusterID == "" {
  841. clusterID = defaultClusterID
  842. }
  843. pvUsedCostMap[clusterID] += result.Values[0].Value
  844. }
  845. }
  846. }
  847. if ctx.HasErrors() {
  848. for _, err := range ctx.Errors() {
  849. log.Errorf("ComputeClusterCosts: %s", err)
  850. }
  851. return nil, ctx.ErrorCollection()
  852. }
  853. // Convert intermediate structure to Costs instances
  854. costsByCluster := map[string]*ClusterCosts{}
  855. for id, cd := range costData {
  856. dataMins, ok := dataMinsByCluster[id]
  857. if !ok {
  858. dataMins = mins
  859. klog.V(3).Infof("[Warning] cluster cost data count not found for cluster %s", id)
  860. }
  861. costs, err := NewClusterCostsFromCumulative(cd["cpu"], cd["gpu"], cd["ram"], cd["storage"]+cd["localstorage"], window, offset, dataMins/timeutil.MinsPerHour)
  862. if err != nil {
  863. klog.V(3).Infof("[Warning] Failed to parse cluster costs on %s (%s) from cumulative data: %+v", window, offset, cd)
  864. return nil, err
  865. }
  866. if cpuBD, ok := cpuBreakdownMap[id]; ok {
  867. costs.CPUBreakdown = cpuBD
  868. }
  869. if ramBD, ok := ramBreakdownMap[id]; ok {
  870. costs.RAMBreakdown = ramBD
  871. }
  872. costs.StorageBreakdown = &ClusterCostsBreakdown{}
  873. if pvUC, ok := pvUsedCostMap[id]; ok {
  874. costs.StorageBreakdown.Idle = (costs.StorageCumulative - pvUC) / costs.StorageCumulative
  875. costs.StorageBreakdown.User = pvUC / costs.StorageCumulative
  876. }
  877. costs.DataMinutes = dataMins
  878. costsByCluster[id] = costs
  879. }
  880. return costsByCluster, nil
  881. }
  882. type Totals struct {
  883. TotalCost [][]string `json:"totalcost"`
  884. CPUCost [][]string `json:"cpucost"`
  885. MemCost [][]string `json:"memcost"`
  886. StorageCost [][]string `json:"storageCost"`
  887. }
  888. func resultToTotals(qrs []*prom.QueryResult) ([][]string, error) {
  889. if len(qrs) == 0 {
  890. return [][]string{}, fmt.Errorf("Not enough data available in the selected time range")
  891. }
  892. result := qrs[0]
  893. totals := [][]string{}
  894. for _, value := range result.Values {
  895. d0 := fmt.Sprintf("%f", value.Timestamp)
  896. d1 := fmt.Sprintf("%f", value.Value)
  897. toAppend := []string{
  898. d0,
  899. d1,
  900. }
  901. totals = append(totals, toAppend)
  902. }
  903. return totals, nil
  904. }
  905. // ClusterCostsOverTime gives the full cluster costs over time
  906. func ClusterCostsOverTime(cli prometheus.Client, provider cloud.Provider, startString, endString string, window, offset time.Duration) (*Totals, error) {
  907. localStorageQuery := provider.GetLocalStorageQuery(window, offset, true, false)
  908. if localStorageQuery != "" {
  909. localStorageQuery = fmt.Sprintf("+ %s", localStorageQuery)
  910. }
  911. layout := "2006-01-02T15:04:05.000Z"
  912. start, err := time.Parse(layout, startString)
  913. if err != nil {
  914. klog.V(1).Infof("Error parsing time %s. Error: %s", startString, err.Error())
  915. return nil, err
  916. }
  917. end, err := time.Parse(layout, endString)
  918. if err != nil {
  919. klog.V(1).Infof("Error parsing time %s. Error: %s", endString, err.Error())
  920. return nil, err
  921. }
  922. fmtWindow := timeutil.DurationString(window)
  923. if fmtWindow == "" {
  924. err := fmt.Errorf("window value invalid or missing")
  925. klog.V(1).Infof("Error parsing time %v. Error: %s", window, err.Error())
  926. return nil, err
  927. }
  928. fmtOffset := timeutil.DurationToPromOffsetString(offset)
  929. qCores := fmt.Sprintf(queryClusterCores, fmtWindow, fmtOffset, env.GetPromClusterLabel(), fmtWindow, fmtOffset, env.GetPromClusterLabel(), fmtWindow, fmtOffset, env.GetPromClusterLabel(), env.GetPromClusterLabel())
  930. qRAM := fmt.Sprintf(queryClusterRAM, fmtWindow, fmtOffset, env.GetPromClusterLabel(), fmtWindow, fmtOffset, env.GetPromClusterLabel(), env.GetPromClusterLabel())
  931. qStorage := fmt.Sprintf(queryStorage, fmtWindow, fmtOffset, env.GetPromClusterLabel(), fmtWindow, fmtOffset, env.GetPromClusterLabel(), env.GetPromClusterLabel(), localStorageQuery)
  932. qTotal := fmt.Sprintf(queryTotal, env.GetPromClusterLabel(), env.GetPromClusterLabel(), env.GetPromClusterLabel(), env.GetPromClusterLabel(), localStorageQuery)
  933. ctx := prom.NewContext(cli)
  934. resChClusterCores := ctx.QueryRange(qCores, start, end, window)
  935. resChClusterRAM := ctx.QueryRange(qRAM, start, end, window)
  936. resChStorage := ctx.QueryRange(qStorage, start, end, window)
  937. resChTotal := ctx.QueryRange(qTotal, start, end, window)
  938. resultClusterCores, err := resChClusterCores.Await()
  939. if err != nil {
  940. return nil, err
  941. }
  942. resultClusterRAM, err := resChClusterRAM.Await()
  943. if err != nil {
  944. return nil, err
  945. }
  946. resultStorage, err := resChStorage.Await()
  947. if err != nil {
  948. return nil, err
  949. }
  950. resultTotal, err := resChTotal.Await()
  951. if err != nil {
  952. return nil, err
  953. }
  954. coreTotal, err := resultToTotals(resultClusterCores)
  955. if err != nil {
  956. klog.Infof("[Warning] ClusterCostsOverTime: no cpu data: %s", err)
  957. return nil, err
  958. }
  959. ramTotal, err := resultToTotals(resultClusterRAM)
  960. if err != nil {
  961. klog.Infof("[Warning] ClusterCostsOverTime: no ram data: %s", err)
  962. return nil, err
  963. }
  964. storageTotal, err := resultToTotals(resultStorage)
  965. if err != nil {
  966. klog.Infof("[Warning] ClusterCostsOverTime: no storage data: %s", err)
  967. }
  968. clusterTotal, err := resultToTotals(resultTotal)
  969. if err != nil {
  970. // If clusterTotal query failed, it's likely because there are no PVs, which
  971. // causes the qTotal query to return no data. Instead, query only node costs.
  972. // If that fails, return an error because something is actually wrong.
  973. qNodes := fmt.Sprintf(queryNodes, env.GetPromClusterLabel(), localStorageQuery)
  974. resultNodes, warnings, err := ctx.QueryRangeSync(qNodes, start, end, window)
  975. for _, warning := range warnings {
  976. log.Warningf(warning)
  977. }
  978. if err != nil {
  979. return nil, err
  980. }
  981. clusterTotal, err = resultToTotals(resultNodes)
  982. if err != nil {
  983. klog.Infof("[Warning] ClusterCostsOverTime: no node data: %s", err)
  984. return nil, err
  985. }
  986. }
  987. return &Totals{
  988. TotalCost: clusterTotal,
  989. CPUCost: coreTotal,
  990. MemCost: ramTotal,
  991. StorageCost: storageTotal,
  992. }, nil
  993. }