allocation.go 52 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504
  1. package costmodel
  2. import (
  3. "fmt"
  4. "time"
  5. "github.com/kubecost/cost-model/pkg/env"
  6. "github.com/kubecost/cost-model/pkg/kubecost"
  7. "github.com/kubecost/cost-model/pkg/log"
  8. "github.com/kubecost/cost-model/pkg/prom"
  9. "github.com/kubecost/cost-model/pkg/thanos"
  10. "k8s.io/apimachinery/pkg/labels"
  11. )
  12. const (
  13. queryFmtMinutes = `avg(kube_pod_container_status_running{}) by (container, pod, namespace, cluster_id)[%s:%s]%s`
  14. queryFmtRAMBytesAllocated = `avg(avg_over_time(container_memory_allocation_bytes{container!="", container!="POD", node!=""}[%s]%s)) by (container, pod, namespace, node, cluster_id)`
  15. queryFmtRAMRequests = `avg(avg_over_time(kube_pod_container_resource_requests_memory_bytes{container!="", container!="POD", node!=""}[%s]%s)) by (container, pod, namespace, node, cluster_id)`
  16. queryFmtRAMUsage = `avg(avg_over_time(container_memory_working_set_bytes{container_name!="", container_name!="POD", instance!=""}[%s]%s)) by (container_name, pod_name, namespace, instance, cluster_id)`
  17. queryFmtCPUCoresAllocated = `avg(avg_over_time(container_cpu_allocation{container!="", container!="POD", node!=""}[%s]%s)) by (container, pod, namespace, node, cluster_id)`
  18. queryFmtCPURequests = `avg(avg_over_time(kube_pod_container_resource_requests_cpu_cores{container!="", container!="POD", node!=""}[%s]%s)) by (container, pod, namespace, node, cluster_id)`
  19. queryFmtCPUUsage = `avg(rate(container_cpu_usage_seconds_total{container_name!="", container_name!="POD", instance!=""}[%s]%s)) by (container_name, pod_name, namespace, instance, cluster_id)`
  20. queryFmtGPUsRequested = `avg(avg_over_time(kube_pod_container_resource_requests{resource="nvidia_com_gpu", container!="",container!="POD", node!=""}[%s]%s)) by (container, pod, namespace, node, cluster_id)`
  21. queryFmtNodeCostPerCPUHr = `avg(avg_over_time(node_cpu_hourly_cost[%s]%s)) by (node, cluster_id, instance_type)`
  22. queryFmtNodeCostPerRAMGiBHr = `avg(avg_over_time(node_ram_hourly_cost[%s]%s)) by (node, cluster_id, instance_type)`
  23. queryFmtNodeCostPerGPUHr = `avg(avg_over_time(node_gpu_hourly_cost[%s]%s)) by (node, cluster_id, instance_type)`
  24. queryFmtNodeIsSpot = `avg_over_time(kubecost_node_is_spot[%s]%s)`
  25. queryFmtPVCInfo = `avg(kube_persistentvolumeclaim_info{volumename != ""}) by (persistentvolumeclaim, storageclass, volumename, namespace, cluster_id)[%s:%s]%s`
  26. queryFmtPVBytes = `avg(avg_over_time(kube_persistentvolume_capacity_bytes[%s]%s)) by (persistentvolume, cluster_id)`
  27. queryFmtPodPVCAllocation = `avg(avg_over_time(pod_pvc_allocation[%s]%s)) by (persistentvolume, persistentvolumeclaim, pod, namespace, cluster_id)`
  28. queryFmtPVCBytesRequested = `avg(avg_over_time(kube_persistentvolumeclaim_resource_requests_storage_bytes{}[%s]%s)) by (persistentvolumeclaim, namespace, cluster_id)`
  29. queryFmtPVCostPerGiBHour = `avg(avg_over_time(pv_hourly_cost[%s]%s)) by (volumename, cluster_id)`
  30. queryFmtNetZoneGiB = `sum(increase(kubecost_pod_network_egress_bytes_total{internet="false", sameZone="false", sameRegion="true"}[%s]%s)) by (pod_name, namespace, cluster_id) / 1024 / 1024 / 1024`
  31. queryFmtNetZoneCostPerGiB = `avg(avg_over_time(kubecost_network_zone_egress_cost{}[%s]%s)) by (cluster_id)`
  32. queryFmtNetRegionGiB = `sum(increase(kubecost_pod_network_egress_bytes_total{internet="false", sameZone="false", sameRegion="false"}[%s]%s)) by (pod_name, namespace, cluster_id) / 1024 / 1024 / 1024`
  33. queryFmtNetRegionCostPerGiB = `avg(avg_over_time(kubecost_network_region_egress_cost{}[%s]%s)) by (cluster_id)`
  34. queryFmtNetInternetGiB = `sum(increase(kubecost_pod_network_egress_bytes_total{internet="true"}[%s]%s)) by (pod_name, namespace, cluster_id) / 1024 / 1024 / 1024`
  35. queryFmtNetInternetCostPerGiB = `avg(avg_over_time(kubecost_network_internet_egress_cost{}[%s]%s)) by (cluster_id)`
  36. queryFmtNamespaceLabels = `avg_over_time(kube_namespace_labels[%s]%s)`
  37. queryFmtNamespaceAnnotations = `avg_over_time(kube_namespace_annotations[%s]%s)`
  38. queryFmtPodLabels = `avg_over_time(kube_pod_labels[%s]%s)`
  39. queryFmtPodAnnotations = `avg_over_time(kube_pod_annotations[%s]%s)`
  40. queryFmtServiceLabels = `avg_over_time(service_selector_labels[%s]%s)`
  41. queryFmtDeploymentLabels = `avg_over_time(deployment_match_labels[%s]%s)`
  42. queryFmtStatefulSetLabels = `avg_over_time(statefulSet_match_labels[%s]%s)`
  43. queryFmtDaemonSetLabels = `sum(avg_over_time(kube_pod_owner{owner_kind="DaemonSet"}[%s]%s)) by (pod, owner_name, namespace, cluster_id)`
  44. queryFmtJobLabels = `sum(avg_over_time(kube_pod_owner{owner_kind="Job"}[%s]%s)) by (pod, owner_name, namespace ,cluster_id)`
  45. )
  46. // TODO niko/computeallocation idle minutes = 1?
  47. // ComputeAllocation uses the CostModel instance to compute an AllocationSet
  48. // for the window defined by the given start and end times. The Allocations
  49. // returned are unaggregated (i.e. down to the container level).
  50. func (cm *CostModel) ComputeAllocation(start, end time.Time) (*kubecost.AllocationSet, error) {
  51. // Create a window spanning the requested query
  52. s, e := start, end
  53. window := kubecost.NewWindow(&s, &e)
  54. // Create an empty AllocationSet. For safety, in the case of an error, we
  55. // should prefer to return this empty set with the error. (In the case of
  56. // no error, of course we populate the set and return it.)
  57. allocSet := kubecost.NewAllocationSet(start, end)
  58. // Convert window (start, end) to (duration, offset) for querying Prometheus
  59. timesToDurations := func(s, e time.Time) (dur, off time.Duration) {
  60. now := time.Now()
  61. off = now.Sub(e)
  62. dur = e.Sub(s)
  63. return dur, off
  64. }
  65. duration, offset := timesToDurations(start, end)
  66. // If using Thanos, increase offset to 3 hours, reducing the duration by
  67. // equal measure to maintain the same starting point.
  68. thanosDur := thanos.OffsetDuration()
  69. if offset < thanosDur && env.IsThanosEnabled() {
  70. diff := thanosDur - offset
  71. offset += diff
  72. duration -= diff
  73. }
  74. // If duration < 0, return an empty set
  75. if duration < 0 {
  76. return allocSet, nil
  77. }
  78. // Negative offset means that the end time is in the future. Prometheus
  79. // fails for non-positive offset values, so shrink the duration and
  80. // remove the offset altogether.
  81. if offset < 0 {
  82. duration = duration + offset
  83. offset = 0
  84. }
  85. durStr := fmt.Sprintf("%dm", int64(duration.Minutes()))
  86. offStr := fmt.Sprintf(" offset %dm", int64(offset.Minutes()))
  87. if offset < time.Minute {
  88. offStr = ""
  89. }
  90. // TODO niko/computeallocation dynamic resolution? add to ComputeAllocation() in allocation.Source?
  91. resStr := "1m"
  92. // resPerHr := 60
  93. startQuerying := time.Now()
  94. ctx := prom.NewContext(cm.PrometheusClient)
  95. // TODO niko/computeallocation retries? (That should probably go into the Store.)
  96. // TODO niko/computeallocation split into required and optional queries?
  97. queryMinutes := fmt.Sprintf(queryFmtMinutes, durStr, resStr, offStr)
  98. resChMinutes := ctx.Query(queryMinutes)
  99. queryRAMBytesAllocated := fmt.Sprintf(queryFmtRAMBytesAllocated, durStr, offStr)
  100. resChRAMBytesAllocated := ctx.Query(queryRAMBytesAllocated)
  101. queryRAMRequests := fmt.Sprintf(queryFmtRAMRequests, durStr, offStr)
  102. resChRAMRequests := ctx.Query(queryRAMRequests)
  103. queryRAMUsage := fmt.Sprintf(queryFmtRAMUsage, durStr, offStr)
  104. resChRAMUsage := ctx.Query(queryRAMUsage)
  105. queryCPUCoresAllocated := fmt.Sprintf(queryFmtCPUCoresAllocated, durStr, offStr)
  106. resChCPUCoresAllocated := ctx.Query(queryCPUCoresAllocated)
  107. queryCPURequests := fmt.Sprintf(queryFmtCPURequests, durStr, offStr)
  108. resChCPURequests := ctx.Query(queryCPURequests)
  109. queryCPUUsage := fmt.Sprintf(queryFmtCPUUsage, durStr, offStr)
  110. resChCPUUsage := ctx.Query(queryCPUUsage)
  111. queryGPUsRequested := fmt.Sprintf(queryFmtGPUsRequested, durStr, offStr)
  112. resChGPUsRequested := ctx.Query(queryGPUsRequested)
  113. queryNodeCostPerCPUHr := fmt.Sprintf(queryFmtNodeCostPerCPUHr, durStr, offStr)
  114. resChNodeCostPerCPUHr := ctx.Query(queryNodeCostPerCPUHr)
  115. queryNodeCostPerRAMGiBHr := fmt.Sprintf(queryFmtNodeCostPerRAMGiBHr, durStr, offStr)
  116. resChNodeCostPerRAMGiBHr := ctx.Query(queryNodeCostPerRAMGiBHr)
  117. queryNodeCostPerGPUHr := fmt.Sprintf(queryFmtNodeCostPerGPUHr, durStr, offStr)
  118. resChNodeCostPerGPUHr := ctx.Query(queryNodeCostPerGPUHr)
  119. queryNodeIsSpot := fmt.Sprintf(queryFmtNodeIsSpot, durStr, offStr)
  120. resChNodeIsSpot := ctx.Query(queryNodeIsSpot)
  121. queryPVCInfo := fmt.Sprintf(queryFmtPVCInfo, durStr, resStr, offStr)
  122. resChPVCInfo := ctx.Query(queryPVCInfo)
  123. queryPVBytes := fmt.Sprintf(queryFmtPVBytes, durStr, offStr)
  124. resChPVBytes := ctx.Query(queryPVBytes)
  125. queryPodPVCAllocation := fmt.Sprintf(queryFmtPodPVCAllocation, durStr, offStr)
  126. resChPodPVCAllocation := ctx.Query(queryPodPVCAllocation)
  127. queryPVCBytesRequested := fmt.Sprintf(queryFmtPVCBytesRequested, durStr, offStr)
  128. resChPVCBytesRequested := ctx.Query(queryPVCBytesRequested)
  129. queryPVCostPerGiBHour := fmt.Sprintf(queryFmtPVCostPerGiBHour, durStr, offStr)
  130. resChPVCostPerGiBHour := ctx.Query(queryPVCostPerGiBHour)
  131. queryNetZoneGiB := fmt.Sprintf(queryFmtNetZoneGiB, durStr, offStr)
  132. resChNetZoneGiB := ctx.Query(queryNetZoneGiB)
  133. queryNetZoneCostPerGiB := fmt.Sprintf(queryFmtNetZoneCostPerGiB, durStr, offStr)
  134. resChNetZoneCostPerGiB := ctx.Query(queryNetZoneCostPerGiB)
  135. queryNetRegionGiB := fmt.Sprintf(queryFmtNetRegionGiB, durStr, offStr)
  136. resChNetRegionGiB := ctx.Query(queryNetRegionGiB)
  137. queryNetRegionCostPerGiB := fmt.Sprintf(queryFmtNetRegionCostPerGiB, durStr, offStr)
  138. resChNetRegionCostPerGiB := ctx.Query(queryNetRegionCostPerGiB)
  139. queryNetInternetGiB := fmt.Sprintf(queryFmtNetInternetGiB, durStr, offStr)
  140. resChNetInternetGiB := ctx.Query(queryNetInternetGiB)
  141. queryNetInternetCostPerGiB := fmt.Sprintf(queryFmtNetInternetCostPerGiB, durStr, offStr)
  142. resChNetInternetCostPerGiB := ctx.Query(queryNetInternetCostPerGiB)
  143. queryNamespaceLabels := fmt.Sprintf(queryFmtNamespaceLabels, durStr, offStr)
  144. resChNamespaceLabels := ctx.Query(queryNamespaceLabels)
  145. queryNamespaceAnnotations := fmt.Sprintf(queryFmtNamespaceAnnotations, durStr, offStr)
  146. resChNamespaceAnnotations := ctx.Query(queryNamespaceAnnotations)
  147. queryPodLabels := fmt.Sprintf(queryFmtPodLabels, durStr, offStr)
  148. resChPodLabels := ctx.Query(queryPodLabels)
  149. queryPodAnnotations := fmt.Sprintf(queryFmtPodAnnotations, durStr, offStr)
  150. resChPodAnnotations := ctx.Query(queryPodAnnotations)
  151. queryServiceLabels := fmt.Sprintf(queryFmtServiceLabels, durStr, offStr)
  152. resChServiceLabels := ctx.Query(queryServiceLabels)
  153. queryDeploymentLabels := fmt.Sprintf(queryFmtDeploymentLabels, durStr, offStr)
  154. resChDeploymentLabels := ctx.Query(queryDeploymentLabels)
  155. queryStatefulSetLabels := fmt.Sprintf(queryFmtStatefulSetLabels, durStr, offStr)
  156. resChStatefulSetLabels := ctx.Query(queryStatefulSetLabels)
  157. queryDaemonSetLabels := fmt.Sprintf(queryFmtDaemonSetLabels, durStr, offStr)
  158. resChDaemonSetLabels := ctx.Query(queryDaemonSetLabels)
  159. queryJobLabels := fmt.Sprintf(queryFmtJobLabels, durStr, offStr)
  160. resChJobLabels := ctx.Query(queryJobLabels)
  161. resMinutes, _ := resChMinutes.Await()
  162. resCPUCoresAllocated, _ := resChCPUCoresAllocated.Await()
  163. resCPURequests, _ := resChCPURequests.Await()
  164. resCPUUsage, _ := resChCPUUsage.Await()
  165. resRAMBytesAllocated, _ := resChRAMBytesAllocated.Await()
  166. resRAMRequests, _ := resChRAMRequests.Await()
  167. resRAMUsage, _ := resChRAMUsage.Await()
  168. resGPUsRequested, _ := resChGPUsRequested.Await()
  169. resNodeCostPerCPUHr, _ := resChNodeCostPerCPUHr.Await()
  170. resNodeCostPerRAMGiBHr, _ := resChNodeCostPerRAMGiBHr.Await()
  171. resNodeCostPerGPUHr, _ := resChNodeCostPerGPUHr.Await()
  172. resNodeIsSpot, _ := resChNodeIsSpot.Await()
  173. resPVBytes, _ := resChPVBytes.Await()
  174. resPVCostPerGiBHour, _ := resChPVCostPerGiBHour.Await()
  175. resPVCInfo, _ := resChPVCInfo.Await()
  176. resPVCBytesRequested, _ := resChPVCBytesRequested.Await()
  177. resPodPVCAllocation, _ := resChPodPVCAllocation.Await()
  178. resNetZoneGiB, _ := resChNetZoneGiB.Await()
  179. resNetZoneCostPerGiB, _ := resChNetZoneCostPerGiB.Await()
  180. resNetRegionGiB, _ := resChNetRegionGiB.Await()
  181. resNetRegionCostPerGiB, _ := resChNetRegionCostPerGiB.Await()
  182. resNetInternetGiB, _ := resChNetInternetGiB.Await()
  183. resNetInternetCostPerGiB, _ := resChNetInternetCostPerGiB.Await()
  184. resNamespaceLabels, _ := resChNamespaceLabels.Await()
  185. resNamespaceAnnotations, _ := resChNamespaceAnnotations.Await()
  186. resPodLabels, _ := resChPodLabels.Await()
  187. resPodAnnotations, _ := resChPodAnnotations.Await()
  188. resServiceLabels, _ := resChServiceLabels.Await()
  189. resDeploymentLabels, _ := resChDeploymentLabels.Await()
  190. resStatefulSetLabels, _ := resChStatefulSetLabels.Await()
  191. resDaemonSetLabels, _ := resChDaemonSetLabels.Await()
  192. resJobLabels, _ := resChJobLabels.Await()
  193. // ---------------------------------------
  194. // TODO niko/computeallocation remove logs
  195. log.Infof("CostModel.ComputeAllocation: %s", queryMinutes)
  196. log.Infof("CostModel.ComputeAllocation: %s", queryRAMBytesAllocated)
  197. log.Infof("CostModel.ComputeAllocation: %s", queryCPUCoresAllocated)
  198. log.Infof("CostModel.ComputeAllocation: %s", queryGPUsRequested)
  199. // ---------------------------------------
  200. log.Profile(startQuerying, "CostModel.ComputeAllocation: queries complete")
  201. defer log.Profile(time.Now(), "CostModel.ComputeAllocation: processing complete")
  202. // Build out a map of Allocations, starting with (start, end) so that we
  203. // begin with minutes, from which we compute resource allocation and cost
  204. // totals from measured rate data.
  205. allocationMap := map[containerKey]*kubecost.Allocation{}
  206. // Keep track of the allocations per pod, for the sake of splitting PVC and
  207. // Network allocation into per-Allocation from per-Pod.
  208. podAllocation := map[podKey][]*kubecost.Allocation{}
  209. // clusterStarts and clusterEnds record the earliest start and latest end
  210. // times, respectively, on a cluster-basis. These are used for unmounted
  211. // PVs and other "virtual" Allocations so that minutes are maximally
  212. // accurate during start-up or spin-down of a cluster
  213. clusterStart := map[string]time.Time{}
  214. clusterEnd := map[string]time.Time{}
  215. buildAllocationMap(window, allocationMap, podAllocation, clusterStart, clusterEnd, resMinutes)
  216. applyCPUCoresAllocated(allocationMap, resCPUCoresAllocated)
  217. applyCPUCoresRequested(allocationMap, resCPURequests)
  218. applyCPUCoresUsed(allocationMap, resCPUUsage)
  219. applyRAMBytesAllocated(allocationMap, resRAMBytesAllocated)
  220. applyRAMBytesRequested(allocationMap, resRAMRequests)
  221. applyRAMBytesUsed(allocationMap, resRAMUsage)
  222. applyGPUsRequested(allocationMap, resGPUsRequested)
  223. applyNetworkAllocation(allocationMap, podAllocation, resNetZoneGiB, resNetZoneCostPerGiB)
  224. applyNetworkAllocation(allocationMap, podAllocation, resNetRegionGiB, resNetRegionCostPerGiB)
  225. applyNetworkAllocation(allocationMap, podAllocation, resNetInternetGiB, resNetInternetCostPerGiB)
  226. // TODO niko/computeallocation pruneDuplicateData? (see costmodel.go)
  227. namespaceLabels := resToNamespaceLabels(resNamespaceLabels)
  228. podLabels := resToPodLabels(resPodLabels)
  229. namespaceAnnotations := resToNamespaceAnnotations(resNamespaceAnnotations)
  230. podAnnotations := resToPodAnnotations(resPodAnnotations)
  231. applyLabels(allocationMap, namespaceLabels, podLabels)
  232. applyAnnotations(allocationMap, namespaceAnnotations, podAnnotations)
  233. serviceLabels := getServiceLabels(resServiceLabels)
  234. applyServicesToPods(allocationMap, podLabels, serviceLabels)
  235. podDeploymentMap := labelsToPodControllerMap(podLabels, resToDeploymentLabels(resDeploymentLabels))
  236. podStatefulSetMap := labelsToPodControllerMap(podLabels, resToStatefulSetLabels(resStatefulSetLabels))
  237. podDaemonSetMap := resToPodDaemonSetMap(resDaemonSetLabels)
  238. podJobMap := resToPodJobMap(resJobLabels)
  239. applyControllersToPods(allocationMap, podDeploymentMap)
  240. applyControllersToPods(allocationMap, podStatefulSetMap)
  241. applyControllersToPods(allocationMap, podDaemonSetMap)
  242. applyControllersToPods(allocationMap, podJobMap)
  243. // TODO breakdown network costs?
  244. // Build out a map of Nodes with resource costs, discounts, and node types
  245. // for converting resource allocation data to cumulative costs.
  246. nodeMap := map[nodeKey]*Node{}
  247. applyNodeCostPerCPUHr(nodeMap, resNodeCostPerCPUHr)
  248. applyNodeCostPerRAMGiBHr(nodeMap, resNodeCostPerRAMGiBHr)
  249. applyNodeCostPerGPUHr(nodeMap, resNodeCostPerGPUHr)
  250. applyNodeSpot(nodeMap, resNodeIsSpot)
  251. applyNodeDiscount(nodeMap, cm)
  252. // Build out the map of all PVs with class, size and cost-per-hour.
  253. // Note: this does not record time running, which we may want to
  254. // include later for increased PV precision. (As long as the PV has
  255. // a PVC, we get time running there, so this is only inaccurate
  256. // for short-lived, unmounted PVs.)
  257. pvMap := map[pvKey]*PV{}
  258. buildPVMap(pvMap, resPVCostPerGiBHour)
  259. applyPVBytes(pvMap, resPVBytes)
  260. // Build out the map of all PVCs with time running, bytes requested,
  261. // and connect to the correct PV from pvMap. (If no PV exists, that
  262. // is noted, but does not result in any allocation/cost.)
  263. pvcMap := map[pvcKey]*PVC{}
  264. buildPVCMap(window, pvcMap, pvMap, resPVCInfo)
  265. applyPVCBytesRequested(pvcMap, resPVCBytesRequested)
  266. // Build out the relationships of pods to their PVCs. This step
  267. // populates the PVC.Count field so that PVC allocation can be
  268. // split appropriately among each pod's container allocation.
  269. podPVCMap := map[podKey][]*PVC{}
  270. buildPodPVCMap(podPVCMap, pvMap, pvcMap, podAllocation, resPodPVCAllocation)
  271. // Identify unmounted PVs (PVs without PVCs) and add one Allocation per
  272. // cluster representing each cluster's unmounted PVs (if necessary).
  273. applyUnmountedPVs(window, allocationMap, pvMap, pvcMap)
  274. for _, alloc := range allocationMap {
  275. cluster, _ := alloc.Properties.GetCluster()
  276. node, _ := alloc.Properties.GetNode()
  277. namespace, _ := alloc.Properties.GetNamespace()
  278. pod, _ := alloc.Properties.GetPod()
  279. container, _ := alloc.Properties.GetContainer()
  280. podKey := newPodKey(cluster, namespace, pod)
  281. nodeKey := newNodeKey(cluster, node)
  282. if n, ok := nodeMap[nodeKey]; !ok {
  283. if pod != kubecost.UnmountedSuffix {
  284. log.Warningf("CostModel.ComputeAllocation: failed to find node %s for %s", nodeKey, alloc.Name)
  285. }
  286. } else {
  287. alloc.CPUCost = alloc.CPUCoreHours * n.CostPerCPUHr
  288. alloc.RAMCost = (alloc.RAMByteHours / 1024 / 1024 / 1024) * n.CostPerRAMGiBHr
  289. alloc.GPUCost = alloc.GPUHours * n.CostPerGPUHr
  290. }
  291. if pvcs, ok := podPVCMap[podKey]; ok {
  292. for _, pvc := range pvcs {
  293. // Determine the (start, end) of the relationship between the
  294. // given PVC and the associated Allocation so that a precise
  295. // number of hours can be used to compute cumulative cost.
  296. s, e := alloc.Start, alloc.End
  297. if pvc.Start.After(alloc.Start) {
  298. s = pvc.Start
  299. }
  300. if pvc.End.Before(alloc.End) {
  301. e = pvc.End
  302. }
  303. minutes := e.Sub(s).Minutes()
  304. hrs := minutes / 60.0
  305. count := float64(pvc.Count)
  306. if pvc.Count < 1 {
  307. count = 1
  308. }
  309. gib := pvc.Bytes / 1024 / 1024 / 1024
  310. cost := pvc.Volume.CostPerGiBHour * gib * hrs
  311. // Apply the size and cost of the PV to the allocation, each
  312. // weighted by count (i.e. the number of containers in the pod)
  313. alloc.PVByteHours += pvc.Bytes * hrs / count
  314. alloc.PVCost += cost / count
  315. }
  316. }
  317. alloc.TotalCost = 0.0
  318. alloc.TotalCost += alloc.CPUCost
  319. alloc.TotalCost += alloc.RAMCost
  320. alloc.TotalCost += alloc.GPUCost
  321. alloc.TotalCost += alloc.PVCost
  322. alloc.TotalCost += alloc.NetworkCost
  323. alloc.TotalCost += alloc.SharedCost
  324. alloc.TotalCost += alloc.ExternalCost
  325. // if alloc.RAMBytesRequestAverage > 0 {
  326. // alloc.RAMEfficiency = alloc.RAMBytesUsageAverage / alloc.RAMBytesRequestAverage
  327. // } else {
  328. // alloc.RAMEfficiency = 1.0
  329. // }
  330. // if alloc.CPUCoreRequestAverage > 0 {
  331. // alloc.CPUEfficiency = alloc.CPUCoreUsageAverage / alloc.CPUCoreRequestAverage
  332. // } else {
  333. // alloc.CPUEfficiency = 1.0
  334. // }
  335. // if alloc.CPUCost+alloc.RAMCost > 0 {
  336. // ramCostEff := alloc.RAMEfficiency * alloc.RAMCost
  337. // cpuCostEff := alloc.CPUEfficiency * alloc.CPUCost
  338. // alloc.TotalEfficiency = (ramCostEff + cpuCostEff) / (alloc.CPUCost + alloc.RAMCost)
  339. // }
  340. // Make sure that the name is correct (node may not be present at this
  341. // point due to it missing from queryMinutes) then insert.
  342. alloc.Name = fmt.Sprintf("%s/%s/%s/%s/%s", cluster, node, namespace, pod, container)
  343. allocSet.Set(alloc)
  344. }
  345. return allocSet, nil
  346. }
  347. func buildAllocationMap(window kubecost.Window, allocationMap map[containerKey]*kubecost.Allocation, podAllocation map[podKey][]*kubecost.Allocation, clusterStart, clusterEnd map[string]time.Time, resMinutes []*prom.QueryResult) {
  348. for _, res := range resMinutes {
  349. if len(res.Values) == 0 {
  350. log.Warningf("CostModel.ComputeAllocation: empty minutes result")
  351. continue
  352. }
  353. cluster, err := res.GetString("cluster_id")
  354. if err != nil {
  355. cluster = env.GetClusterID()
  356. }
  357. labels, err := res.GetStrings("namespace", "pod", "container")
  358. if err != nil {
  359. log.Warningf("CostModel.ComputeAllocation: minutes query result missing field: %s", err)
  360. continue
  361. }
  362. namespace := labels["namespace"]
  363. pod := labels["pod"]
  364. container := labels["container"]
  365. containerKey := newContainerKey(cluster, namespace, pod, container)
  366. podKey := newPodKey(cluster, namespace, pod)
  367. // allocStart and allocEnd are the timestamps of the first and last
  368. // minutes the allocation was running, respectively. We subtract 1m
  369. // from allocStart because this point will actually represent the end
  370. // of the first minute. We don't subtract from allocEnd because it
  371. // already represents the end of the last minute.
  372. var allocStart, allocEnd time.Time
  373. for _, datum := range res.Values {
  374. t := time.Unix(int64(datum.Timestamp), 0)
  375. if allocStart.IsZero() && datum.Value > 0 && window.Contains(t) {
  376. allocStart = t
  377. }
  378. if datum.Value > 0 && window.Contains(t) {
  379. allocEnd = t
  380. }
  381. }
  382. if allocStart.IsZero() || allocEnd.IsZero() {
  383. continue
  384. }
  385. allocStart = allocStart.Add(-time.Minute)
  386. // Set start if unset or this datum's start time is earlier than the
  387. // current earliest time.
  388. if _, ok := clusterStart[cluster]; !ok || allocStart.Before(clusterStart[cluster]) {
  389. clusterStart[cluster] = allocStart
  390. }
  391. // Set end if unset or this datum's end time is later than the
  392. // current latest time.
  393. if _, ok := clusterEnd[cluster]; !ok || allocEnd.After(clusterEnd[cluster]) {
  394. clusterEnd[cluster] = allocEnd
  395. }
  396. name := fmt.Sprintf("%s/%s/%s/%s", cluster, namespace, pod, container)
  397. alloc := &kubecost.Allocation{
  398. Name: name,
  399. Properties: kubecost.Properties{},
  400. Window: window.Clone(),
  401. Start: allocStart,
  402. End: allocEnd,
  403. }
  404. alloc.Properties.SetContainer(container)
  405. alloc.Properties.SetPod(pod)
  406. alloc.Properties.SetNamespace(namespace)
  407. alloc.Properties.SetCluster(cluster)
  408. allocationMap[containerKey] = alloc
  409. if _, ok := podAllocation[podKey]; !ok {
  410. podAllocation[podKey] = []*kubecost.Allocation{}
  411. }
  412. podAllocation[podKey] = append(podAllocation[podKey], alloc)
  413. }
  414. }
  415. func applyCPUCoresAllocated(allocationMap map[containerKey]*kubecost.Allocation, resCPUCoresAllocated []*prom.QueryResult) {
  416. for _, res := range resCPUCoresAllocated {
  417. key, err := resultContainerKey(res, "cluster_id", "namespace", "pod", "container")
  418. if err != nil {
  419. log.Warningf("CostModel.ComputeAllocation: CPU allocation query result missing field: %s", err)
  420. continue
  421. }
  422. _, ok := allocationMap[key]
  423. if !ok {
  424. // TODO niko/computeallocation dedupe log
  425. log.Warningf("CostModel.ComputeAllocation: unidentified CPU allocation query result: %s", key)
  426. continue
  427. }
  428. cpuCores := res.Values[0].Value
  429. hours := allocationMap[key].Minutes() / 60.0
  430. allocationMap[key].CPUCoreHours = cpuCores * hours
  431. node, err := res.GetString("node")
  432. if err != nil {
  433. log.Warningf("CostModel.ComputeAllocation: CPU allocation query result missing 'node': %s", key)
  434. continue
  435. }
  436. allocationMap[key].Properties.SetNode(node)
  437. }
  438. }
  439. func applyCPUCoresRequested(allocationMap map[containerKey]*kubecost.Allocation, resCPUCoresRequested []*prom.QueryResult) {
  440. for _, res := range resCPUCoresRequested {
  441. key, err := resultContainerKey(res, "cluster_id", "namespace", "pod", "container")
  442. if err != nil {
  443. log.Warningf("CostModel.ComputeAllocation: CPU request query result missing field: %s", err)
  444. continue
  445. }
  446. _, ok := allocationMap[key]
  447. if !ok {
  448. continue
  449. }
  450. allocationMap[key].CPUCoreRequestAverage = res.Values[0].Value
  451. // If CPU allocation is less than requests, set CPUCoreHours to
  452. // request level.
  453. if allocationMap[key].CPUCores() < res.Values[0].Value {
  454. allocationMap[key].CPUCoreHours = res.Values[0].Value * (allocationMap[key].Minutes() / 60.0)
  455. }
  456. node, err := res.GetString("node")
  457. if err != nil {
  458. log.Warningf("CostModel.ComputeAllocation: CPU request query result missing 'node': %s", key)
  459. continue
  460. }
  461. allocationMap[key].Properties.SetNode(node)
  462. }
  463. }
  464. func applyCPUCoresUsed(allocationMap map[containerKey]*kubecost.Allocation, resCPUCoresUsed []*prom.QueryResult) {
  465. for _, res := range resCPUCoresUsed {
  466. key, err := resultContainerKey(res, "cluster_id", "namespace", "pod_name", "container_name")
  467. if err != nil {
  468. log.Warningf("CostModel.ComputeAllocation: CPU usage query result missing field: %s", err)
  469. continue
  470. }
  471. _, ok := allocationMap[key]
  472. if !ok {
  473. // TODO niko/computeallocation dedupe log
  474. log.Warningf("CostModel.ComputeAllocation: unidentified CPU usage query result: %s", key)
  475. continue
  476. }
  477. allocationMap[key].CPUCoreUsageAverage = res.Values[0].Value
  478. }
  479. }
  480. func applyRAMBytesAllocated(allocationMap map[containerKey]*kubecost.Allocation, resRAMBytesAllocated []*prom.QueryResult) {
  481. for _, res := range resRAMBytesAllocated {
  482. key, err := resultContainerKey(res, "cluster_id", "namespace", "pod", "container")
  483. if err != nil {
  484. log.Warningf("CostModel.ComputeAllocation: RAM allocation query result missing field: %s", err)
  485. continue
  486. }
  487. _, ok := allocationMap[key]
  488. if !ok {
  489. // TODO niko/computeallocation dedupe log
  490. log.Warningf("CostModel.ComputeAllocation: unidentified RAM allocation query result: %s", key)
  491. continue
  492. }
  493. ramBytes := res.Values[0].Value
  494. hours := allocationMap[key].Minutes() / 60.0
  495. allocationMap[key].RAMByteHours = ramBytes * hours
  496. node, err := res.GetString("node")
  497. if err != nil {
  498. log.Warningf("CostModel.ComputeAllocation: RAM allocation query result missing 'node': %s", key)
  499. continue
  500. }
  501. allocationMap[key].Properties.SetNode(node)
  502. }
  503. }
  504. func applyRAMBytesRequested(allocationMap map[containerKey]*kubecost.Allocation, resRAMBytesRequested []*prom.QueryResult) {
  505. for _, res := range resRAMBytesRequested {
  506. key, err := resultContainerKey(res, "cluster_id", "namespace", "pod", "container")
  507. if err != nil {
  508. log.Warningf("CostModel.ComputeAllocation: RAM request query result missing field: %s", err)
  509. continue
  510. }
  511. _, ok := allocationMap[key]
  512. if !ok {
  513. continue
  514. }
  515. allocationMap[key].RAMBytesRequestAverage = res.Values[0].Value
  516. // If RAM allocation is less than requests, set RAMByteHours to
  517. // request level.
  518. if allocationMap[key].RAMBytes() < res.Values[0].Value {
  519. allocationMap[key].RAMByteHours = res.Values[0].Value * (allocationMap[key].Minutes() / 60.0)
  520. }
  521. node, err := res.GetString("node")
  522. if err != nil {
  523. log.Warningf("CostModel.ComputeAllocation: RAM request query result missing 'node': %s", key)
  524. continue
  525. }
  526. allocationMap[key].Properties.SetNode(node)
  527. }
  528. }
  529. func applyRAMBytesUsed(allocationMap map[containerKey]*kubecost.Allocation, resRAMBytesUsed []*prom.QueryResult) {
  530. for _, res := range resRAMBytesUsed {
  531. key, err := resultContainerKey(res, "cluster_id", "namespace", "pod_name", "container_name")
  532. if err != nil {
  533. log.Warningf("CostModel.ComputeAllocation: RAM usage query result missing field: %s", err)
  534. continue
  535. }
  536. _, ok := allocationMap[key]
  537. if !ok {
  538. // TODO niko/computeallocation dedupe log
  539. log.Warningf("CostModel.ComputeAllocation: unidentified RAM usage query result: %s", key)
  540. continue
  541. }
  542. allocationMap[key].RAMBytesUsageAverage = res.Values[0].Value
  543. }
  544. }
  545. func applyGPUsRequested(allocationMap map[containerKey]*kubecost.Allocation, resGPUsRequested []*prom.QueryResult) {
  546. for _, res := range resGPUsRequested {
  547. key, err := resultContainerKey(res, "cluster_id", "namespace", "pod", "container")
  548. if err != nil {
  549. log.Warningf("CostModel.ComputeAllocation: GPU allocation query result missing field: %s", err)
  550. continue
  551. }
  552. _, ok := allocationMap[key]
  553. if !ok {
  554. // TODO niko/computeallocation dedupe log
  555. log.Warningf("CostModel.ComputeAllocation: unidentified GPU allocation query result: %s", key)
  556. continue
  557. }
  558. // TODO niko/computeallocation remove log
  559. log.Infof("CostModel.ComputeAllocation: GPU results: %s=%f", key, res.Values[0].Value)
  560. hrs := allocationMap[key].Minutes() / 60.0
  561. allocationMap[key].GPUHours = res.Values[0].Value * hrs
  562. }
  563. }
  564. func applyNetworkAllocation(allocationMap map[containerKey]*kubecost.Allocation, podAllocation map[podKey][]*kubecost.Allocation, resNetworkGiB []*prom.QueryResult, resNetworkCostPerGiB []*prom.QueryResult) {
  565. costPerGiBByCluster := map[string]float64{}
  566. for _, res := range resNetworkCostPerGiB {
  567. cluster, err := res.GetString("cluster_id")
  568. if err != nil {
  569. cluster = env.GetClusterID()
  570. }
  571. costPerGiBByCluster[cluster] = res.Values[0].Value
  572. }
  573. for _, res := range resNetworkGiB {
  574. podKey, err := resultPodKey(res, "cluster_id", "namespace", "pod_name")
  575. if err != nil {
  576. log.Warningf("CostModel.ComputeAllocation: Network allocation query result missing field: %s", err)
  577. continue
  578. }
  579. allocs, ok := podAllocation[podKey]
  580. if !ok {
  581. log.Warningf("CostModel.ComputeAllocation: Network allocation query result for unidentified pod allocations: %s", podKey)
  582. continue
  583. }
  584. for _, alloc := range allocs {
  585. gib := res.Values[0].Value
  586. costPerGiB := costPerGiBByCluster[podKey.Cluster]
  587. alloc.NetworkCost = gib * costPerGiB
  588. }
  589. }
  590. }
  591. func resToNamespaceLabels(resNamespaceLabels []*prom.QueryResult) map[string]map[string]string {
  592. namespaceLabels := map[string]map[string]string{}
  593. for _, res := range resNamespaceLabels {
  594. namespace, err := res.GetString("namespace")
  595. if err != nil {
  596. continue
  597. }
  598. if _, ok := namespaceLabels[namespace]; !ok {
  599. namespaceLabels[namespace] = map[string]string{}
  600. }
  601. for k, l := range res.GetLabels() {
  602. namespaceLabels[namespace][k] = l
  603. }
  604. }
  605. return namespaceLabels
  606. }
  607. func resToPodLabels(resPodLabels []*prom.QueryResult) map[podKey]map[string]string {
  608. podLabels := map[podKey]map[string]string{}
  609. for _, res := range resPodLabels {
  610. podKey, err := resultPodKey(res, "cluster_id", "namespace", "pod")
  611. if err != nil {
  612. continue
  613. }
  614. if _, ok := podLabels[podKey]; !ok {
  615. podLabels[podKey] = map[string]string{}
  616. }
  617. for k, l := range res.GetLabels() {
  618. podLabels[podKey][k] = l
  619. }
  620. }
  621. return podLabels
  622. }
  623. func resToNamespaceAnnotations(resNamespaceAnnotations []*prom.QueryResult) map[string]map[string]string {
  624. namespaceAnnotations := map[string]map[string]string{}
  625. for _, res := range resNamespaceAnnotations {
  626. namespace, err := res.GetString("namespace")
  627. if err != nil {
  628. continue
  629. }
  630. if _, ok := namespaceAnnotations[namespace]; !ok {
  631. namespaceAnnotations[namespace] = map[string]string{}
  632. }
  633. for k, l := range res.GetAnnotations() {
  634. namespaceAnnotations[namespace][k] = l
  635. }
  636. }
  637. return namespaceAnnotations
  638. }
  639. func resToPodAnnotations(resPodAnnotations []*prom.QueryResult) map[podKey]map[string]string {
  640. podAnnotations := map[podKey]map[string]string{}
  641. for _, res := range resPodAnnotations {
  642. podKey, err := resultPodKey(res, "cluster_id", "namespace", "pod")
  643. if err != nil {
  644. continue
  645. }
  646. if _, ok := podAnnotations[podKey]; !ok {
  647. podAnnotations[podKey] = map[string]string{}
  648. }
  649. for k, l := range res.GetAnnotations() {
  650. podAnnotations[podKey][k] = l
  651. }
  652. }
  653. return podAnnotations
  654. }
  655. func applyLabels(allocationMap map[containerKey]*kubecost.Allocation, namespaceLabels map[string]map[string]string, podLabels map[podKey]map[string]string) {
  656. for key, alloc := range allocationMap {
  657. allocLabels, err := alloc.Properties.GetLabels()
  658. if err != nil {
  659. allocLabels = map[string]string{}
  660. }
  661. // Apply namespace labels first, then pod labels so that pod labels
  662. // overwrite namespace labels.
  663. if labels, ok := namespaceLabels[key.Namespace]; ok {
  664. for k, v := range labels {
  665. allocLabels[k] = v
  666. }
  667. }
  668. podKey := newPodKey(key.Cluster, key.Namespace, key.Pod)
  669. if labels, ok := podLabels[podKey]; ok {
  670. for k, v := range labels {
  671. allocLabels[k] = v
  672. }
  673. }
  674. alloc.Properties.SetLabels(allocLabels)
  675. }
  676. }
  677. func applyAnnotations(allocationMap map[containerKey]*kubecost.Allocation, namespaceAnnotations map[string]map[string]string, podAnnotations map[podKey]map[string]string) {
  678. for key, alloc := range allocationMap {
  679. allocAnnotations, err := alloc.Properties.GetAnnotations()
  680. if err != nil {
  681. allocAnnotations = map[string]string{}
  682. }
  683. // Apply namespace annotations first, then pod annotations so that
  684. // pod labels overwrite namespace labels.
  685. if labels, ok := namespaceAnnotations[key.Namespace]; ok {
  686. for k, v := range labels {
  687. allocAnnotations[k] = v
  688. }
  689. }
  690. podKey := newPodKey(key.Cluster, key.Namespace, key.Pod)
  691. if labels, ok := podAnnotations[podKey]; ok {
  692. for k, v := range labels {
  693. allocAnnotations[k] = v
  694. }
  695. }
  696. alloc.Properties.SetAnnotations(allocAnnotations)
  697. }
  698. }
  699. func getServiceLabels(resServiceLabels []*prom.QueryResult) map[serviceKey]map[string]string {
  700. serviceLabels := map[serviceKey]map[string]string{}
  701. for _, res := range resServiceLabels {
  702. serviceKey, err := resultServiceKey(res, "cluster_id", "namespace", "service")
  703. if err != nil {
  704. continue
  705. }
  706. if _, ok := serviceLabels[serviceKey]; !ok {
  707. serviceLabels[serviceKey] = map[string]string{}
  708. }
  709. for k, l := range res.GetLabels() {
  710. serviceLabels[serviceKey][k] = l
  711. }
  712. }
  713. return serviceLabels
  714. }
  715. func resToDeploymentLabels(resDeploymentLabels []*prom.QueryResult) map[controllerKey]map[string]string {
  716. deploymentLabels := map[controllerKey]map[string]string{}
  717. for _, res := range resDeploymentLabels {
  718. controllerKey, err := resultDeploymentKey(res, "cluster_id", "namespace", "deployment")
  719. if err != nil {
  720. continue
  721. }
  722. if _, ok := deploymentLabels[controllerKey]; !ok {
  723. deploymentLabels[controllerKey] = map[string]string{}
  724. }
  725. for k, l := range res.GetLabels() {
  726. deploymentLabels[controllerKey][k] = l
  727. }
  728. }
  729. return deploymentLabels
  730. }
  731. func resToStatefulSetLabels(resStatefulSetLabels []*prom.QueryResult) map[controllerKey]map[string]string {
  732. statefulSetLabels := map[controllerKey]map[string]string{}
  733. for _, res := range resStatefulSetLabels {
  734. controllerKey, err := resultStatefulSetKey(res, "cluster_id", "namespace", "statefulSet")
  735. if err != nil {
  736. continue
  737. }
  738. if _, ok := statefulSetLabels[controllerKey]; !ok {
  739. statefulSetLabels[controllerKey] = map[string]string{}
  740. }
  741. for k, l := range res.GetLabels() {
  742. statefulSetLabels[controllerKey][k] = l
  743. }
  744. }
  745. return statefulSetLabels
  746. }
  747. func labelsToPodControllerMap(podLabels map[podKey]map[string]string, controllerLabels map[controllerKey]map[string]string) map[podKey]controllerKey {
  748. podControllerMap := map[podKey]controllerKey{}
  749. // For each controller, turn the labels into a selector and attempt to
  750. // match it with each set of pod labels. A match indicates that the pod
  751. // belongs to the controller.
  752. for cKey, cLabels := range controllerLabels {
  753. selector := labels.Set(cLabels).AsSelectorPreValidated()
  754. for pKey, pLabels := range podLabels {
  755. // If the pod is in a different cluster or namespace, there is
  756. // no need to compare the labels.
  757. if cKey.Cluster != pKey.Cluster || cKey.Namespace != pKey.Namespace {
  758. continue
  759. }
  760. podLabelSet := labels.Set(pLabels)
  761. if selector.Matches(podLabelSet) {
  762. if _, ok := podControllerMap[pKey]; ok {
  763. log.Warningf("CostModel.ComputeAllocation: PodControllerMap match already exists: %s matches %s and %s", pKey, podControllerMap[pKey], cKey)
  764. }
  765. podControllerMap[pKey] = cKey
  766. }
  767. }
  768. }
  769. return podControllerMap
  770. }
  771. func resToPodDaemonSetMap(resDaemonSetLabels []*prom.QueryResult) map[podKey]controllerKey {
  772. daemonSetLabels := map[podKey]controllerKey{}
  773. for _, res := range resDaemonSetLabels {
  774. controllerKey, err := resultDaemonSetKey(res, "cluster_id", "namespace", "owner_name")
  775. if err != nil {
  776. continue
  777. }
  778. pod, err := res.GetString("pod")
  779. if err != nil {
  780. log.Warningf("CostModel.ComputeAllocation: DaemonSetLabel result without pod: %s", controllerKey)
  781. }
  782. podKey := newPodKey(controllerKey.Cluster, controllerKey.Namespace, pod)
  783. daemonSetLabels[podKey] = controllerKey
  784. }
  785. return daemonSetLabels
  786. }
  787. func resToPodJobMap(resJobLabels []*prom.QueryResult) map[podKey]controllerKey {
  788. jobLabels := map[podKey]controllerKey{}
  789. for _, res := range resJobLabels {
  790. controllerKey, err := resultJobKey(res, "cluster_id", "namespace", "owner_name")
  791. if err != nil {
  792. continue
  793. }
  794. pod, err := res.GetString("pod")
  795. if err != nil {
  796. log.Warningf("CostModel.ComputeAllocation: JobLabel result without pod: %s", controllerKey)
  797. }
  798. podKey := newPodKey(controllerKey.Cluster, controllerKey.Namespace, pod)
  799. jobLabels[podKey] = controllerKey
  800. }
  801. return jobLabels
  802. }
  803. func applyServicesToPods(allocationMap map[containerKey]*kubecost.Allocation, podLabels map[podKey]map[string]string, serviceLabels map[serviceKey]map[string]string) {
  804. podServicesMap := map[podKey][]serviceKey{}
  805. // For each service, turn the labels into a selector and attempt to
  806. // match it with each set of pod labels. A match indicates that the pod
  807. // belongs to the service.
  808. for sKey, sLabels := range serviceLabels {
  809. selector := labels.Set(sLabels).AsSelectorPreValidated()
  810. for pKey, pLabels := range podLabels {
  811. // If the pod is in a different cluster or namespace, there is
  812. // no need to compare the labels.
  813. if sKey.Cluster != pKey.Cluster || sKey.Namespace != pKey.Namespace {
  814. continue
  815. }
  816. podLabelSet := labels.Set(pLabels)
  817. if selector.Matches(podLabelSet) {
  818. if _, ok := podServicesMap[pKey]; !ok {
  819. podServicesMap[pKey] = []serviceKey{}
  820. }
  821. podServicesMap[pKey] = append(podServicesMap[pKey], sKey)
  822. }
  823. }
  824. }
  825. // For each allocation, attempt to find and apply the list of services
  826. // associated with the allocation's pod.
  827. for key, alloc := range allocationMap {
  828. pKey := newPodKey(key.Cluster, key.Namespace, key.Pod)
  829. if sKeys, ok := podServicesMap[pKey]; ok {
  830. services := []string{}
  831. for _, sKey := range sKeys {
  832. services = append(services, sKey.Service)
  833. }
  834. alloc.Properties.SetServices(services)
  835. }
  836. }
  837. }
  838. func applyControllersToPods(allocationMap map[containerKey]*kubecost.Allocation, podControllerMap map[podKey]controllerKey) {
  839. for key, alloc := range allocationMap {
  840. podKey := newPodKey(key.Cluster, key.Namespace, key.Pod)
  841. if controllerKey, ok := podControllerMap[podKey]; ok {
  842. alloc.Properties.SetControllerKind(controllerKey.ControllerKind)
  843. alloc.Properties.SetController(controllerKey.Controller)
  844. }
  845. }
  846. }
  847. func applyNodeCostPerCPUHr(nodeMap map[nodeKey]*Node, resNodeCostPerCPUHr []*prom.QueryResult) {
  848. for _, res := range resNodeCostPerCPUHr {
  849. cluster, err := res.GetString("cluster_id")
  850. if err != nil {
  851. cluster = env.GetClusterID()
  852. }
  853. node, err := res.GetString("node")
  854. if err != nil {
  855. log.Warningf("CostModel.ComputeAllocation: Node CPU cost query result missing field: %s", err)
  856. continue
  857. }
  858. instanceType, err := res.GetString("instance_type")
  859. if err != nil {
  860. log.Warningf("CostModel.ComputeAllocation: Node CPU cost query result missing field: %s", err)
  861. continue
  862. }
  863. key := newNodeKey(cluster, node)
  864. if _, ok := nodeMap[key]; !ok {
  865. nodeMap[key] = &Node{
  866. Name: node,
  867. NodeType: instanceType,
  868. }
  869. }
  870. nodeMap[key].CostPerCPUHr = res.Values[0].Value
  871. }
  872. }
  873. func applyNodeCostPerRAMGiBHr(nodeMap map[nodeKey]*Node, resNodeCostPerRAMGiBHr []*prom.QueryResult) {
  874. for _, res := range resNodeCostPerRAMGiBHr {
  875. cluster, err := res.GetString("cluster_id")
  876. if err != nil {
  877. cluster = env.GetClusterID()
  878. }
  879. node, err := res.GetString("node")
  880. if err != nil {
  881. log.Warningf("CostModel.ComputeAllocation: Node RAM cost query result missing field: %s", err)
  882. continue
  883. }
  884. instanceType, err := res.GetString("instance_type")
  885. if err != nil {
  886. log.Warningf("CostModel.ComputeAllocation: Node RAM cost query result missing field: %s", err)
  887. continue
  888. }
  889. key := newNodeKey(cluster, node)
  890. if _, ok := nodeMap[key]; !ok {
  891. nodeMap[key] = &Node{
  892. Name: node,
  893. NodeType: instanceType,
  894. }
  895. }
  896. nodeMap[key].CostPerRAMGiBHr = res.Values[0].Value
  897. }
  898. }
  899. func applyNodeCostPerGPUHr(nodeMap map[nodeKey]*Node, resNodeCostPerGPUHr []*prom.QueryResult) {
  900. for _, res := range resNodeCostPerGPUHr {
  901. cluster, err := res.GetString("cluster_id")
  902. if err != nil {
  903. cluster = env.GetClusterID()
  904. }
  905. node, err := res.GetString("node")
  906. if err != nil {
  907. log.Warningf("CostModel.ComputeAllocation: Node GPU cost query result missing field: %s", err)
  908. continue
  909. }
  910. instanceType, err := res.GetString("instance_type")
  911. if err != nil {
  912. log.Warningf("CostModel.ComputeAllocation: Node GPU cost query result missing field: %s", err)
  913. continue
  914. }
  915. key := newNodeKey(cluster, node)
  916. if _, ok := nodeMap[key]; !ok {
  917. nodeMap[key] = &Node{
  918. Name: node,
  919. NodeType: instanceType,
  920. }
  921. }
  922. nodeMap[key].CostPerGPUHr = res.Values[0].Value
  923. }
  924. }
  925. func applyNodeSpot(nodeMap map[nodeKey]*Node, resNodeIsSpot []*prom.QueryResult) {
  926. for _, res := range resNodeIsSpot {
  927. cluster, err := res.GetString("cluster_id")
  928. if err != nil {
  929. cluster = env.GetClusterID()
  930. }
  931. node, err := res.GetString("node")
  932. if err != nil {
  933. log.Warningf("CostModel.ComputeAllocation: Node spot query result missing field: %s", err)
  934. continue
  935. }
  936. key := newNodeKey(cluster, node)
  937. if _, ok := nodeMap[key]; !ok {
  938. log.Warningf("CostModel.ComputeAllocation: Node spot query result for missing node: %s", key)
  939. continue
  940. }
  941. nodeMap[key].Preemptible = res.Values[0].Value > 0
  942. }
  943. }
  944. func applyNodeDiscount(nodeMap map[nodeKey]*Node, cm *CostModel) {
  945. if cm == nil {
  946. return
  947. }
  948. c, err := cm.Provider.GetConfig()
  949. if err != nil {
  950. log.Errorf("CostModel.ComputeAllocation: applyNodeDiscount: %s", err)
  951. return
  952. }
  953. discount, err := ParsePercentString(c.Discount)
  954. if err != nil {
  955. log.Errorf("CostModel.ComputeAllocation: applyNodeDiscount: %s", err)
  956. return
  957. }
  958. negotiatedDiscount, err := ParsePercentString(c.NegotiatedDiscount)
  959. if err != nil {
  960. log.Errorf("CostModel.ComputeAllocation: applyNodeDiscount: %s", err)
  961. return
  962. }
  963. for _, node := range nodeMap {
  964. // TODO niko/computeallocation GKE Reserved Instances into account
  965. node.Discount = cm.Provider.CombinedDiscountForNode(node.NodeType, node.Preemptible, discount, negotiatedDiscount)
  966. node.CostPerCPUHr *= (1.0 - node.Discount)
  967. node.CostPerRAMGiBHr *= (1.0 - node.Discount)
  968. }
  969. }
  970. func buildPVMap(pvMap map[pvKey]*PV, resPVCostPerGiBHour []*prom.QueryResult) {
  971. for _, res := range resPVCostPerGiBHour {
  972. cluster, err := res.GetString("cluster_id")
  973. if err != nil {
  974. cluster = env.GetClusterID()
  975. }
  976. name, err := res.GetString("volumename")
  977. if err != nil {
  978. log.Warningf("CostModel.ComputeAllocation: PV cost without volumename")
  979. continue
  980. }
  981. key := newPVKey(cluster, name)
  982. pvMap[key] = &PV{
  983. Cluster: cluster,
  984. Name: name,
  985. CostPerGiBHour: res.Values[0].Value,
  986. }
  987. }
  988. }
  989. func applyPVBytes(pvMap map[pvKey]*PV, resPVBytes []*prom.QueryResult) {
  990. for _, res := range resPVBytes {
  991. key, err := resultPVKey(res, "cluster_id", "persistentvolume")
  992. if err != nil {
  993. log.Warningf("CostModel.ComputeAllocation: PV bytes query result missing field: %s", err)
  994. continue
  995. }
  996. if _, ok := pvMap[key]; !ok {
  997. log.Warningf("CostModel.ComputeAllocation: PV bytes result for missing PV: %s", err)
  998. continue
  999. }
  1000. pvMap[key].Bytes = res.Values[0].Value
  1001. }
  1002. }
  1003. func buildPVCMap(window kubecost.Window, pvcMap map[pvcKey]*PVC, pvMap map[pvKey]*PV, resPVCInfo []*prom.QueryResult) {
  1004. for _, res := range resPVCInfo {
  1005. cluster, err := res.GetString("cluster_id")
  1006. if err != nil {
  1007. cluster = env.GetClusterID()
  1008. }
  1009. values, err := res.GetStrings("persistentvolumeclaim", "storageclass", "volumename", "namespace")
  1010. if err != nil {
  1011. log.Warningf("CostModel.ComputeAllocation: PVC info query result missing field: %s", err)
  1012. continue
  1013. }
  1014. namespace := values["namespace"]
  1015. name := values["persistentvolumeclaim"]
  1016. volume := values["volumename"]
  1017. storageClass := values["storageclass"]
  1018. pvKey := newPVKey(cluster, volume)
  1019. pvcKey := newPVCKey(cluster, namespace, name)
  1020. // pvcStart and pvcEnd are the timestamps of the first and last minutes
  1021. // the PVC was running, respectively. We subtract 1m from pvcStart
  1022. // because this point will actually represent the end of the first
  1023. // minute. We don't subtract from pvcEnd because it already represents
  1024. // the end of the last minute.
  1025. var pvcStart, pvcEnd time.Time
  1026. for _, datum := range res.Values {
  1027. t := time.Unix(int64(datum.Timestamp), 0)
  1028. if pvcStart.IsZero() && datum.Value > 0 && window.Contains(t) {
  1029. pvcStart = t
  1030. }
  1031. if datum.Value > 0 && window.Contains(t) {
  1032. pvcEnd = t
  1033. }
  1034. }
  1035. if pvcStart.IsZero() || pvcEnd.IsZero() {
  1036. log.Warningf("CostModel.ComputeAllocation: PVC %s has no running time", pvcKey)
  1037. }
  1038. pvcStart = pvcStart.Add(-time.Minute)
  1039. if _, ok := pvMap[pvKey]; !ok {
  1040. log.Warningf("CostModel.ComputeAllocation: PV missing for PVC info query result: %s", pvKey)
  1041. continue
  1042. }
  1043. pvMap[pvKey].StorageClass = storageClass
  1044. if _, ok := pvcMap[pvcKey]; !ok {
  1045. pvcMap[pvcKey] = &PVC{}
  1046. }
  1047. pvcMap[pvcKey].Name = name
  1048. pvcMap[pvcKey].Namespace = namespace
  1049. pvcMap[pvcKey].Volume = pvMap[pvKey]
  1050. pvcMap[pvcKey].Start = pvcStart
  1051. pvcMap[pvcKey].End = pvcEnd
  1052. }
  1053. }
  1054. func applyPVCBytesRequested(pvcMap map[pvcKey]*PVC, resPVCBytesRequested []*prom.QueryResult) {
  1055. for _, res := range resPVCBytesRequested {
  1056. key, err := resultPVCKey(res, "cluster_id", "namespace", "persistentvolumeclaim")
  1057. if err != nil {
  1058. log.Warningf("CostModel.ComputeAllocation: PVC bytes requested query result missing field: %s", err)
  1059. continue
  1060. }
  1061. if _, ok := pvcMap[key]; !ok {
  1062. log.Warningf("CostModel.ComputeAllocation: PVC bytes requested result for missing PVC: %s", key)
  1063. continue
  1064. }
  1065. pvcMap[key].Bytes = res.Values[0].Value
  1066. }
  1067. }
  1068. func buildPodPVCMap(podPVCMap map[podKey][]*PVC, pvMap map[pvKey]*PV, pvcMap map[pvcKey]*PVC, podAllocation map[podKey][]*kubecost.Allocation, resPodPVCAllocation []*prom.QueryResult) {
  1069. for _, res := range resPodPVCAllocation {
  1070. cluster, err := res.GetString("cluster_id")
  1071. if err != nil {
  1072. cluster = env.GetClusterID()
  1073. }
  1074. values, err := res.GetStrings("persistentvolume", "persistentvolumeclaim", "pod", "namespace")
  1075. if err != nil {
  1076. log.Warningf("CostModel.ComputeAllocation: PVC allocation query result missing field: %s", err)
  1077. continue
  1078. }
  1079. namespace := values["namespace"]
  1080. pod := values["pod"]
  1081. name := values["persistentvolumeclaim"]
  1082. volume := values["persistentvolume"]
  1083. podKey := newPodKey(cluster, namespace, pod)
  1084. pvKey := newPVKey(cluster, volume)
  1085. pvcKey := newPVCKey(cluster, namespace, name)
  1086. if _, ok := pvMap[pvKey]; !ok {
  1087. log.Warningf("CostModel.ComputeAllocation: PV missing for PVC allocation query result: %s", pvKey)
  1088. continue
  1089. }
  1090. if _, ok := podPVCMap[podKey]; !ok {
  1091. podPVCMap[podKey] = []*PVC{}
  1092. }
  1093. pvc, ok := pvcMap[pvcKey]
  1094. if !ok {
  1095. log.Warningf("CostModel.ComputeAllocation: PVC missing for PVC allocation query: %s", pvcKey)
  1096. continue
  1097. }
  1098. pvc.Count = len(podAllocation[podKey])
  1099. pvc.Mounted = true
  1100. podPVCMap[podKey] = append(podPVCMap[podKey], pvc)
  1101. }
  1102. }
  1103. func applyUnmountedPVs(window kubecost.Window, allocationMap map[containerKey]*kubecost.Allocation, pvMap map[pvKey]*PV, pvcMap map[pvcKey]*PVC) {
  1104. unmountedPVBytes := map[string]float64{}
  1105. unmountedPVCost := map[string]float64{}
  1106. for _, pv := range pvMap {
  1107. mounted := false
  1108. for _, pvc := range pvcMap {
  1109. if pvc.Volume == nil {
  1110. continue
  1111. }
  1112. if pvc.Volume == pv {
  1113. mounted = true
  1114. break
  1115. }
  1116. }
  1117. if !mounted {
  1118. gib := pv.Bytes / 1024 / 1024 / 1024
  1119. hrs := window.Minutes() / 60.0 // TODO niko/computeallocation PV hours, not window hours?
  1120. cost := pv.CostPerGiBHour * gib * hrs
  1121. unmountedPVCost[pv.Cluster] += cost
  1122. unmountedPVBytes[pv.Cluster] += pv.Bytes
  1123. }
  1124. }
  1125. for cluster, amount := range unmountedPVCost {
  1126. container := kubecost.UnmountedSuffix
  1127. pod := kubecost.UnmountedSuffix
  1128. namespace := kubecost.UnmountedSuffix
  1129. node := ""
  1130. containerKey := newContainerKey(cluster, namespace, pod, container)
  1131. allocationMap[containerKey] = &kubecost.Allocation{
  1132. Name: fmt.Sprintf("%s/%s/%s/%s/%s", cluster, node, namespace, pod, container),
  1133. Properties: kubecost.Properties{
  1134. kubecost.ClusterProp: cluster,
  1135. kubecost.NodeProp: node,
  1136. kubecost.NamespaceProp: namespace,
  1137. kubecost.PodProp: pod,
  1138. kubecost.ContainerProp: container,
  1139. },
  1140. Window: window.Clone(),
  1141. Start: *window.Start(),
  1142. End: *window.End(),
  1143. PVByteHours: unmountedPVBytes[cluster] * window.Minutes() / 60.0,
  1144. PVCost: amount,
  1145. TotalCost: amount,
  1146. }
  1147. }
  1148. }
  1149. func applyUnmountedPVCs(window kubecost.Window, allocationMap map[containerKey]*kubecost.Allocation, pvcMap map[pvcKey]*PVC) {
  1150. unmountedPVCBytes := map[namespaceKey]float64{}
  1151. unmountedPVCCost := map[namespaceKey]float64{}
  1152. for _, pvc := range pvcMap {
  1153. if !pvc.Mounted && pvc.Volume != nil {
  1154. key := newNamespaceKey(pvc.Cluster, pvc.Namespace)
  1155. gib := pvc.Volume.Bytes / 1024 / 1024 / 1024
  1156. hrs := pvc.Minutes() / 60.0
  1157. cost := pvc.Volume.CostPerGiBHour * gib * hrs
  1158. unmountedPVCCost[key] += cost
  1159. unmountedPVCBytes[key] += pvc.Volume.Bytes
  1160. }
  1161. }
  1162. for key, amount := range unmountedPVCCost {
  1163. container := kubecost.UnmountedSuffix
  1164. pod := kubecost.UnmountedSuffix
  1165. namespace := key.Namespace
  1166. node := ""
  1167. cluster := key.Cluster
  1168. containerKey := newContainerKey(cluster, namespace, pod, container)
  1169. allocationMap[containerKey] = &kubecost.Allocation{
  1170. Name: fmt.Sprintf("%s/%s/%s/%s/%s", cluster, node, namespace, pod, container),
  1171. Properties: kubecost.Properties{
  1172. kubecost.ClusterProp: cluster,
  1173. kubecost.NodeProp: node,
  1174. kubecost.NamespaceProp: namespace,
  1175. kubecost.PodProp: pod,
  1176. kubecost.ContainerProp: container,
  1177. },
  1178. Window: window.Clone(),
  1179. Start: *window.Start(),
  1180. End: *window.End(),
  1181. PVByteHours: unmountedPVCBytes[key] * window.Minutes() / 60.0,
  1182. PVCost: amount,
  1183. TotalCost: amount,
  1184. }
  1185. }
  1186. }
  1187. // PVC describes a PersistentVolumeClaim
  1188. // TODO move to pkg/kubecost? [TODO:CLEANUP]
  1189. // TODO add PersistentVolumeClaims field to type Allocation? [TODO:CLEANUP]
  1190. type PVC struct {
  1191. Bytes float64 `json:"bytes"`
  1192. Count int `json:"count"`
  1193. Name string `json:"name"`
  1194. Cluster string `json:"cluster"`
  1195. Namespace string `json:"namespace"`
  1196. Volume *PV `json:"persistentVolume"`
  1197. Mounted bool `json:"mounted"`
  1198. Start time.Time `json:"start"`
  1199. End time.Time `json:"end"`
  1200. }
  1201. // Cost computes the cumulative cost of the PVC
  1202. func (pvc *PVC) Cost() float64 {
  1203. if pvc == nil || pvc.Volume == nil {
  1204. return 0.0
  1205. }
  1206. gib := pvc.Bytes / 1024 / 1024 / 1024
  1207. hrs := pvc.Minutes() / 60.0
  1208. return pvc.Volume.CostPerGiBHour * gib * hrs
  1209. }
  1210. // Minutes computes the number of minutes over which the PVC is defined
  1211. func (pvc *PVC) Minutes() float64 {
  1212. if pvc == nil {
  1213. return 0.0
  1214. }
  1215. return pvc.End.Sub(pvc.Start).Minutes()
  1216. }
  1217. // String returns a string representation of the PVC
  1218. func (pvc *PVC) String() string {
  1219. if pvc == nil {
  1220. return "<nil>"
  1221. }
  1222. return fmt.Sprintf("%s/%s/%s{Bytes:%.2f, Cost:%.6f, Start,End:%s}", pvc.Cluster, pvc.Namespace, pvc.Name, pvc.Bytes, pvc.Cost(), kubecost.NewWindow(&pvc.Start, &pvc.End))
  1223. }
  1224. // PV describes a PersistentVolume
  1225. // TODO move to pkg/kubecost? [TODO:CLEANUP]
  1226. type PV struct {
  1227. Bytes float64 `json:"bytes"`
  1228. CostPerGiBHour float64 `json:"costPerGiBHour"` // TODO niko/computeallocation GiB or GB?
  1229. Cluster string `json:"cluster"`
  1230. Name string `json:"name"`
  1231. StorageClass string `json:"storageClass"`
  1232. }
  1233. // String returns a string representation of the PV
  1234. func (pv *PV) String() string {
  1235. if pv == nil {
  1236. return "<nil>"
  1237. }
  1238. return fmt.Sprintf("%s/%s{Bytes:%.2f, Cost/GiB*Hr:%.6f, StorageClass:%s}", pv.Cluster, pv.Name, pv.Bytes, pv.CostPerGiBHour, pv.StorageClass)
  1239. }