allocation.go 52 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500
  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 != "unmounted-pvs" {
  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. }
  328. if alloc.CPUCoreRequestAverage > 0 {
  329. alloc.CPUEfficiency = alloc.CPUCoreUsageAverage / alloc.CPUCoreRequestAverage
  330. }
  331. if alloc.CPUCost+alloc.RAMCost > 0 {
  332. ramCostEff := alloc.RAMEfficiency * alloc.RAMCost
  333. cpuCostEff := alloc.CPUEfficiency * alloc.CPUCost
  334. alloc.TotalEfficiency = (ramCostEff + cpuCostEff) / (alloc.CPUCost + alloc.RAMCost)
  335. }
  336. // Make sure that the name is correct (node may not be present at this
  337. // point due to it missing from queryMinutes) then insert.
  338. alloc.Name = fmt.Sprintf("%s/%s/%s/%s/%s", cluster, node, namespace, pod, container)
  339. allocSet.Set(alloc)
  340. }
  341. return allocSet, nil
  342. }
  343. func buildAllocationMap(window kubecost.Window, allocationMap map[containerKey]*kubecost.Allocation, podAllocation map[podKey][]*kubecost.Allocation, clusterStart, clusterEnd map[string]time.Time, resMinutes []*prom.QueryResult) {
  344. for _, res := range resMinutes {
  345. if len(res.Values) == 0 {
  346. log.Warningf("CostModel.ComputeAllocation: empty minutes result")
  347. continue
  348. }
  349. cluster, err := res.GetString("cluster_id")
  350. if err != nil {
  351. cluster = env.GetClusterID()
  352. }
  353. labels, err := res.GetStrings("namespace", "pod", "container")
  354. if err != nil {
  355. log.Warningf("CostModel.ComputeAllocation: minutes query result missing field: %s", err)
  356. continue
  357. }
  358. namespace := labels["namespace"]
  359. pod := labels["pod"]
  360. container := labels["container"]
  361. containerKey := newContainerKey(cluster, namespace, pod, container)
  362. podKey := newPodKey(cluster, namespace, pod)
  363. // allocStart and allocEnd are the timestamps of the first and last
  364. // minutes the allocation was running, respectively. We subtract 1m
  365. // from allocStart because this point will actually represent the end
  366. // of the first minute. We don't subtract from allocEnd because it
  367. // already represents the end of the last minute.
  368. var allocStart, allocEnd time.Time
  369. for _, datum := range res.Values {
  370. t := time.Unix(int64(datum.Timestamp), 0)
  371. if allocStart.IsZero() && datum.Value > 0 && window.Contains(t) {
  372. allocStart = t
  373. }
  374. if datum.Value > 0 && window.Contains(t) {
  375. allocEnd = t
  376. }
  377. }
  378. if allocStart.IsZero() || allocEnd.IsZero() {
  379. continue
  380. }
  381. allocStart = allocStart.Add(-time.Minute)
  382. // Set start if unset or this datum's start time is earlier than the
  383. // current earliest time.
  384. if _, ok := clusterStart[cluster]; !ok || allocStart.Before(clusterStart[cluster]) {
  385. clusterStart[cluster] = allocStart
  386. }
  387. // Set end if unset or this datum's end time is later than the
  388. // current latest time.
  389. if _, ok := clusterEnd[cluster]; !ok || allocEnd.After(clusterEnd[cluster]) {
  390. clusterEnd[cluster] = allocEnd
  391. }
  392. name := fmt.Sprintf("%s/%s/%s/%s", cluster, namespace, pod, container)
  393. alloc := &kubecost.Allocation{
  394. Name: name,
  395. Properties: kubecost.Properties{},
  396. Window: window.Clone(),
  397. Start: allocStart,
  398. End: allocEnd,
  399. }
  400. alloc.Properties.SetContainer(container)
  401. alloc.Properties.SetPod(pod)
  402. alloc.Properties.SetNamespace(namespace)
  403. alloc.Properties.SetCluster(cluster)
  404. allocationMap[containerKey] = alloc
  405. if _, ok := podAllocation[podKey]; !ok {
  406. podAllocation[podKey] = []*kubecost.Allocation{}
  407. }
  408. podAllocation[podKey] = append(podAllocation[podKey], alloc)
  409. }
  410. }
  411. func applyCPUCoresAllocated(allocationMap map[containerKey]*kubecost.Allocation, resCPUCoresAllocated []*prom.QueryResult) {
  412. for _, res := range resCPUCoresAllocated {
  413. key, err := resultContainerKey(res, "cluster_id", "namespace", "pod", "container")
  414. if err != nil {
  415. log.Warningf("CostModel.ComputeAllocation: CPU allocation query result missing field: %s", err)
  416. continue
  417. }
  418. _, ok := allocationMap[key]
  419. if !ok {
  420. // TODO niko/computeallocation dedupe log
  421. log.Warningf("CostModel.ComputeAllocation: unidentified CPU allocation query result: %s", key)
  422. continue
  423. }
  424. cpuCores := res.Values[0].Value
  425. hours := allocationMap[key].Minutes() / 60.0
  426. allocationMap[key].CPUCoreHours = cpuCores * hours
  427. node, err := res.GetString("node")
  428. if err != nil {
  429. log.Warningf("CostModel.ComputeAllocation: CPU allocation query result missing 'node': %s", key)
  430. continue
  431. }
  432. allocationMap[key].Properties.SetNode(node)
  433. }
  434. }
  435. func applyCPUCoresRequested(allocationMap map[containerKey]*kubecost.Allocation, resCPUCoresRequested []*prom.QueryResult) {
  436. for _, res := range resCPUCoresRequested {
  437. key, err := resultContainerKey(res, "cluster_id", "namespace", "pod", "container")
  438. if err != nil {
  439. log.Warningf("CostModel.ComputeAllocation: CPU request query result missing field: %s", err)
  440. continue
  441. }
  442. _, ok := allocationMap[key]
  443. if !ok {
  444. continue
  445. }
  446. allocationMap[key].CPUCoreRequestAverage = res.Values[0].Value
  447. // If CPU allocation is less than requests, set CPUCoreHours to
  448. // request level.
  449. if allocationMap[key].CPUCores() < res.Values[0].Value {
  450. allocationMap[key].CPUCoreHours = res.Values[0].Value * (allocationMap[key].Minutes() / 60.0)
  451. }
  452. node, err := res.GetString("node")
  453. if err != nil {
  454. log.Warningf("CostModel.ComputeAllocation: CPU request query result missing 'node': %s", key)
  455. continue
  456. }
  457. allocationMap[key].Properties.SetNode(node)
  458. }
  459. }
  460. func applyCPUCoresUsed(allocationMap map[containerKey]*kubecost.Allocation, resCPUCoresUsed []*prom.QueryResult) {
  461. for _, res := range resCPUCoresUsed {
  462. key, err := resultContainerKey(res, "cluster_id", "namespace", "pod_name", "container_name")
  463. if err != nil {
  464. log.Warningf("CostModel.ComputeAllocation: CPU usage query result missing field: %s", err)
  465. continue
  466. }
  467. _, ok := allocationMap[key]
  468. if !ok {
  469. // TODO niko/computeallocation dedupe log
  470. log.Warningf("CostModel.ComputeAllocation: unidentified CPU usage query result: %s", key)
  471. continue
  472. }
  473. allocationMap[key].CPUCoreUsageAverage = res.Values[0].Value
  474. }
  475. }
  476. func applyRAMBytesAllocated(allocationMap map[containerKey]*kubecost.Allocation, resRAMBytesAllocated []*prom.QueryResult) {
  477. for _, res := range resRAMBytesAllocated {
  478. key, err := resultContainerKey(res, "cluster_id", "namespace", "pod", "container")
  479. if err != nil {
  480. log.Warningf("CostModel.ComputeAllocation: RAM allocation query result missing field: %s", err)
  481. continue
  482. }
  483. _, ok := allocationMap[key]
  484. if !ok {
  485. // TODO niko/computeallocation dedupe log
  486. log.Warningf("CostModel.ComputeAllocation: unidentified RAM allocation query result: %s", key)
  487. continue
  488. }
  489. ramBytes := res.Values[0].Value
  490. hours := allocationMap[key].Minutes() / 60.0
  491. allocationMap[key].RAMByteHours = ramBytes * hours
  492. node, err := res.GetString("node")
  493. if err != nil {
  494. log.Warningf("CostModel.ComputeAllocation: RAM allocation query result missing 'node': %s", key)
  495. continue
  496. }
  497. allocationMap[key].Properties.SetNode(node)
  498. }
  499. }
  500. func applyRAMBytesRequested(allocationMap map[containerKey]*kubecost.Allocation, resRAMBytesRequested []*prom.QueryResult) {
  501. for _, res := range resRAMBytesRequested {
  502. key, err := resultContainerKey(res, "cluster_id", "namespace", "pod", "container")
  503. if err != nil {
  504. log.Warningf("CostModel.ComputeAllocation: RAM request query result missing field: %s", err)
  505. continue
  506. }
  507. _, ok := allocationMap[key]
  508. if !ok {
  509. continue
  510. }
  511. allocationMap[key].RAMBytesRequestAverage = res.Values[0].Value
  512. // If RAM allocation is less than requests, set RAMByteHours to
  513. // request level.
  514. if allocationMap[key].RAMBytes() < res.Values[0].Value {
  515. allocationMap[key].RAMByteHours = res.Values[0].Value * (allocationMap[key].Minutes() / 60.0)
  516. }
  517. node, err := res.GetString("node")
  518. if err != nil {
  519. log.Warningf("CostModel.ComputeAllocation: RAM request query result missing 'node': %s", key)
  520. continue
  521. }
  522. allocationMap[key].Properties.SetNode(node)
  523. }
  524. }
  525. func applyRAMBytesUsed(allocationMap map[containerKey]*kubecost.Allocation, resRAMBytesUsed []*prom.QueryResult) {
  526. for _, res := range resRAMBytesUsed {
  527. key, err := resultContainerKey(res, "cluster_id", "namespace", "pod_name", "container_name")
  528. if err != nil {
  529. log.Warningf("CostModel.ComputeAllocation: RAM usage query result missing field: %s", err)
  530. continue
  531. }
  532. _, ok := allocationMap[key]
  533. if !ok {
  534. // TODO niko/computeallocation dedupe log
  535. log.Warningf("CostModel.ComputeAllocation: unidentified RAM usage query result: %s", key)
  536. continue
  537. }
  538. allocationMap[key].RAMBytesUsageAverage = res.Values[0].Value
  539. }
  540. }
  541. func applyGPUsRequested(allocationMap map[containerKey]*kubecost.Allocation, resGPUsRequested []*prom.QueryResult) {
  542. for _, res := range resGPUsRequested {
  543. key, err := resultContainerKey(res, "cluster_id", "namespace", "pod", "container")
  544. if err != nil {
  545. log.Warningf("CostModel.ComputeAllocation: GPU allocation query result missing field: %s", err)
  546. continue
  547. }
  548. _, ok := allocationMap[key]
  549. if !ok {
  550. // TODO niko/computeallocation dedupe log
  551. log.Warningf("CostModel.ComputeAllocation: unidentified GPU allocation query result: %s", key)
  552. continue
  553. }
  554. // TODO niko/computeallocation remove log
  555. log.Infof("CostModel.ComputeAllocation: GPU results: %s=%f", key, res.Values[0].Value)
  556. hrs := allocationMap[key].Minutes() / 60.0
  557. allocationMap[key].GPUHours = res.Values[0].Value * hrs
  558. }
  559. }
  560. func applyNetworkAllocation(allocationMap map[containerKey]*kubecost.Allocation, podAllocation map[podKey][]*kubecost.Allocation, resNetworkGiB []*prom.QueryResult, resNetworkCostPerGiB []*prom.QueryResult) {
  561. costPerGiBByCluster := map[string]float64{}
  562. for _, res := range resNetworkCostPerGiB {
  563. cluster, err := res.GetString("cluster_id")
  564. if err != nil {
  565. cluster = env.GetClusterID()
  566. }
  567. costPerGiBByCluster[cluster] = res.Values[0].Value
  568. }
  569. for _, res := range resNetworkGiB {
  570. podKey, err := resultPodKey(res, "cluster_id", "namespace", "pod_name")
  571. if err != nil {
  572. log.Warningf("CostModel.ComputeAllocation: Network allocation query result missing field: %s", err)
  573. continue
  574. }
  575. allocs, ok := podAllocation[podKey]
  576. if !ok {
  577. log.Warningf("CostModel.ComputeAllocation: Network allocation query result for unidentified pod allocations: %s", podKey)
  578. continue
  579. }
  580. for _, alloc := range allocs {
  581. gib := res.Values[0].Value
  582. costPerGiB := costPerGiBByCluster[podKey.Cluster]
  583. alloc.NetworkCost = gib * costPerGiB
  584. }
  585. }
  586. }
  587. func resToNamespaceLabels(resNamespaceLabels []*prom.QueryResult) map[string]map[string]string {
  588. namespaceLabels := map[string]map[string]string{}
  589. for _, res := range resNamespaceLabels {
  590. namespace, err := res.GetString("namespace")
  591. if err != nil {
  592. continue
  593. }
  594. if _, ok := namespaceLabels[namespace]; !ok {
  595. namespaceLabels[namespace] = map[string]string{}
  596. }
  597. for k, l := range res.GetLabels() {
  598. namespaceLabels[namespace][k] = l
  599. }
  600. }
  601. return namespaceLabels
  602. }
  603. func resToPodLabels(resPodLabels []*prom.QueryResult) map[podKey]map[string]string {
  604. podLabels := map[podKey]map[string]string{}
  605. for _, res := range resPodLabels {
  606. podKey, err := resultPodKey(res, "cluster_id", "namespace", "pod")
  607. if err != nil {
  608. continue
  609. }
  610. if _, ok := podLabels[podKey]; !ok {
  611. podLabels[podKey] = map[string]string{}
  612. }
  613. for k, l := range res.GetLabels() {
  614. podLabels[podKey][k] = l
  615. }
  616. }
  617. return podLabels
  618. }
  619. func resToNamespaceAnnotations(resNamespaceAnnotations []*prom.QueryResult) map[string]map[string]string {
  620. namespaceAnnotations := map[string]map[string]string{}
  621. for _, res := range resNamespaceAnnotations {
  622. namespace, err := res.GetString("namespace")
  623. if err != nil {
  624. continue
  625. }
  626. if _, ok := namespaceAnnotations[namespace]; !ok {
  627. namespaceAnnotations[namespace] = map[string]string{}
  628. }
  629. for k, l := range res.GetAnnotations() {
  630. namespaceAnnotations[namespace][k] = l
  631. }
  632. }
  633. return namespaceAnnotations
  634. }
  635. func resToPodAnnotations(resPodAnnotations []*prom.QueryResult) map[podKey]map[string]string {
  636. podAnnotations := map[podKey]map[string]string{}
  637. for _, res := range resPodAnnotations {
  638. podKey, err := resultPodKey(res, "cluster_id", "namespace", "pod")
  639. if err != nil {
  640. continue
  641. }
  642. if _, ok := podAnnotations[podKey]; !ok {
  643. podAnnotations[podKey] = map[string]string{}
  644. }
  645. for k, l := range res.GetAnnotations() {
  646. podAnnotations[podKey][k] = l
  647. }
  648. }
  649. return podAnnotations
  650. }
  651. func applyLabels(allocationMap map[containerKey]*kubecost.Allocation, namespaceLabels map[string]map[string]string, podLabels map[podKey]map[string]string) {
  652. for key, alloc := range allocationMap {
  653. allocLabels, err := alloc.Properties.GetLabels()
  654. if err != nil {
  655. allocLabels = map[string]string{}
  656. }
  657. // Apply namespace labels first, then pod labels so that pod labels
  658. // overwrite namespace labels.
  659. if labels, ok := namespaceLabels[key.Namespace]; ok {
  660. for k, v := range labels {
  661. allocLabels[k] = v
  662. }
  663. }
  664. podKey := newPodKey(key.Cluster, key.Namespace, key.Pod)
  665. if labels, ok := podLabels[podKey]; ok {
  666. for k, v := range labels {
  667. allocLabels[k] = v
  668. }
  669. }
  670. alloc.Properties.SetLabels(allocLabels)
  671. }
  672. }
  673. func applyAnnotations(allocationMap map[containerKey]*kubecost.Allocation, namespaceAnnotations map[string]map[string]string, podAnnotations map[podKey]map[string]string) {
  674. for key, alloc := range allocationMap {
  675. allocAnnotations, err := alloc.Properties.GetAnnotations()
  676. if err != nil {
  677. allocAnnotations = map[string]string{}
  678. }
  679. // Apply namespace annotations first, then pod annotations so that
  680. // pod labels overwrite namespace labels.
  681. if labels, ok := namespaceAnnotations[key.Namespace]; ok {
  682. for k, v := range labels {
  683. allocAnnotations[k] = v
  684. }
  685. }
  686. podKey := newPodKey(key.Cluster, key.Namespace, key.Pod)
  687. if labels, ok := podAnnotations[podKey]; ok {
  688. for k, v := range labels {
  689. allocAnnotations[k] = v
  690. }
  691. }
  692. alloc.Properties.SetAnnotations(allocAnnotations)
  693. }
  694. }
  695. func getServiceLabels(resServiceLabels []*prom.QueryResult) map[serviceKey]map[string]string {
  696. serviceLabels := map[serviceKey]map[string]string{}
  697. for _, res := range resServiceLabels {
  698. serviceKey, err := resultServiceKey(res, "cluster_id", "namespace", "service")
  699. if err != nil {
  700. continue
  701. }
  702. if _, ok := serviceLabels[serviceKey]; !ok {
  703. serviceLabels[serviceKey] = map[string]string{}
  704. }
  705. for k, l := range res.GetLabels() {
  706. serviceLabels[serviceKey][k] = l
  707. }
  708. }
  709. return serviceLabels
  710. }
  711. func resToDeploymentLabels(resDeploymentLabels []*prom.QueryResult) map[controllerKey]map[string]string {
  712. deploymentLabels := map[controllerKey]map[string]string{}
  713. for _, res := range resDeploymentLabels {
  714. controllerKey, err := resultDeploymentKey(res, "cluster_id", "namespace", "deployment")
  715. if err != nil {
  716. continue
  717. }
  718. if _, ok := deploymentLabels[controllerKey]; !ok {
  719. deploymentLabels[controllerKey] = map[string]string{}
  720. }
  721. for k, l := range res.GetLabels() {
  722. deploymentLabels[controllerKey][k] = l
  723. }
  724. }
  725. return deploymentLabels
  726. }
  727. func resToStatefulSetLabels(resStatefulSetLabels []*prom.QueryResult) map[controllerKey]map[string]string {
  728. statefulSetLabels := map[controllerKey]map[string]string{}
  729. for _, res := range resStatefulSetLabels {
  730. controllerKey, err := resultStatefulSetKey(res, "cluster_id", "namespace", "statefulSet")
  731. if err != nil {
  732. continue
  733. }
  734. if _, ok := statefulSetLabels[controllerKey]; !ok {
  735. statefulSetLabels[controllerKey] = map[string]string{}
  736. }
  737. for k, l := range res.GetLabels() {
  738. statefulSetLabels[controllerKey][k] = l
  739. }
  740. }
  741. return statefulSetLabels
  742. }
  743. func labelsToPodControllerMap(podLabels map[podKey]map[string]string, controllerLabels map[controllerKey]map[string]string) map[podKey]controllerKey {
  744. podControllerMap := map[podKey]controllerKey{}
  745. // For each controller, turn the labels into a selector and attempt to
  746. // match it with each set of pod labels. A match indicates that the pod
  747. // belongs to the controller.
  748. for cKey, cLabels := range controllerLabels {
  749. selector := labels.Set(cLabels).AsSelectorPreValidated()
  750. for pKey, pLabels := range podLabels {
  751. // If the pod is in a different cluster or namespace, there is
  752. // no need to compare the labels.
  753. if cKey.Cluster != pKey.Cluster || cKey.Namespace != pKey.Namespace {
  754. continue
  755. }
  756. podLabelSet := labels.Set(pLabels)
  757. if selector.Matches(podLabelSet) {
  758. if _, ok := podControllerMap[pKey]; ok {
  759. log.Warningf("CostModel.ComputeAllocation: PodControllerMap match already exists: %s matches %s and %s", pKey, podControllerMap[pKey], cKey)
  760. }
  761. podControllerMap[pKey] = cKey
  762. }
  763. }
  764. }
  765. return podControllerMap
  766. }
  767. func resToPodDaemonSetMap(resDaemonSetLabels []*prom.QueryResult) map[podKey]controllerKey {
  768. daemonSetLabels := map[podKey]controllerKey{}
  769. for _, res := range resDaemonSetLabels {
  770. controllerKey, err := resultDaemonSetKey(res, "cluster_id", "namespace", "owner_name")
  771. if err != nil {
  772. continue
  773. }
  774. pod, err := res.GetString("pod")
  775. if err != nil {
  776. log.Warningf("CostModel.ComputeAllocation: DaemonSetLabel result without pod: %s", controllerKey)
  777. }
  778. podKey := newPodKey(controllerKey.Cluster, controllerKey.Namespace, pod)
  779. daemonSetLabels[podKey] = controllerKey
  780. }
  781. return daemonSetLabels
  782. }
  783. func resToPodJobMap(resJobLabels []*prom.QueryResult) map[podKey]controllerKey {
  784. jobLabels := map[podKey]controllerKey{}
  785. for _, res := range resJobLabels {
  786. controllerKey, err := resultJobKey(res, "cluster_id", "namespace", "owner_name")
  787. if err != nil {
  788. continue
  789. }
  790. pod, err := res.GetString("pod")
  791. if err != nil {
  792. log.Warningf("CostModel.ComputeAllocation: JobLabel result without pod: %s", controllerKey)
  793. }
  794. podKey := newPodKey(controllerKey.Cluster, controllerKey.Namespace, pod)
  795. jobLabels[podKey] = controllerKey
  796. }
  797. return jobLabels
  798. }
  799. func applyServicesToPods(allocationMap map[containerKey]*kubecost.Allocation, podLabels map[podKey]map[string]string, serviceLabels map[serviceKey]map[string]string) {
  800. podServicesMap := map[podKey][]serviceKey{}
  801. // For each service, turn the labels into a selector and attempt to
  802. // match it with each set of pod labels. A match indicates that the pod
  803. // belongs to the service.
  804. for sKey, sLabels := range serviceLabels {
  805. selector := labels.Set(sLabels).AsSelectorPreValidated()
  806. for pKey, pLabels := range podLabels {
  807. // If the pod is in a different cluster or namespace, there is
  808. // no need to compare the labels.
  809. if sKey.Cluster != pKey.Cluster || sKey.Namespace != pKey.Namespace {
  810. continue
  811. }
  812. podLabelSet := labels.Set(pLabels)
  813. if selector.Matches(podLabelSet) {
  814. if _, ok := podServicesMap[pKey]; !ok {
  815. podServicesMap[pKey] = []serviceKey{}
  816. }
  817. podServicesMap[pKey] = append(podServicesMap[pKey], sKey)
  818. }
  819. }
  820. }
  821. // For each allocation, attempt to find and apply the list of services
  822. // associated with the allocation's pod.
  823. for key, alloc := range allocationMap {
  824. pKey := newPodKey(key.Cluster, key.Namespace, key.Pod)
  825. if sKeys, ok := podServicesMap[pKey]; ok {
  826. services := []string{}
  827. for _, sKey := range sKeys {
  828. services = append(services, sKey.Service)
  829. }
  830. alloc.Properties.SetServices(services)
  831. }
  832. }
  833. }
  834. func applyControllersToPods(allocationMap map[containerKey]*kubecost.Allocation, podControllerMap map[podKey]controllerKey) {
  835. for key, alloc := range allocationMap {
  836. podKey := newPodKey(key.Cluster, key.Namespace, key.Pod)
  837. if controllerKey, ok := podControllerMap[podKey]; ok {
  838. alloc.Properties.SetControllerKind(controllerKey.ControllerKind)
  839. alloc.Properties.SetController(controllerKey.Controller)
  840. }
  841. }
  842. }
  843. func applyNodeCostPerCPUHr(nodeMap map[nodeKey]*Node, resNodeCostPerCPUHr []*prom.QueryResult) {
  844. for _, res := range resNodeCostPerCPUHr {
  845. cluster, err := res.GetString("cluster_id")
  846. if err != nil {
  847. cluster = env.GetClusterID()
  848. }
  849. node, err := res.GetString("node")
  850. if err != nil {
  851. log.Warningf("CostModel.ComputeAllocation: Node CPU cost query result missing field: %s", err)
  852. continue
  853. }
  854. instanceType, err := res.GetString("instance_type")
  855. if err != nil {
  856. log.Warningf("CostModel.ComputeAllocation: Node CPU cost query result missing field: %s", err)
  857. continue
  858. }
  859. key := newNodeKey(cluster, node)
  860. if _, ok := nodeMap[key]; !ok {
  861. nodeMap[key] = &Node{
  862. Name: node,
  863. NodeType: instanceType,
  864. }
  865. }
  866. nodeMap[key].CostPerCPUHr = res.Values[0].Value
  867. }
  868. }
  869. func applyNodeCostPerRAMGiBHr(nodeMap map[nodeKey]*Node, resNodeCostPerRAMGiBHr []*prom.QueryResult) {
  870. for _, res := range resNodeCostPerRAMGiBHr {
  871. cluster, err := res.GetString("cluster_id")
  872. if err != nil {
  873. cluster = env.GetClusterID()
  874. }
  875. node, err := res.GetString("node")
  876. if err != nil {
  877. log.Warningf("CostModel.ComputeAllocation: Node RAM cost query result missing field: %s", err)
  878. continue
  879. }
  880. instanceType, err := res.GetString("instance_type")
  881. if err != nil {
  882. log.Warningf("CostModel.ComputeAllocation: Node RAM cost query result missing field: %s", err)
  883. continue
  884. }
  885. key := newNodeKey(cluster, node)
  886. if _, ok := nodeMap[key]; !ok {
  887. nodeMap[key] = &Node{
  888. Name: node,
  889. NodeType: instanceType,
  890. }
  891. }
  892. nodeMap[key].CostPerRAMGiBHr = res.Values[0].Value
  893. }
  894. }
  895. func applyNodeCostPerGPUHr(nodeMap map[nodeKey]*Node, resNodeCostPerGPUHr []*prom.QueryResult) {
  896. for _, res := range resNodeCostPerGPUHr {
  897. cluster, err := res.GetString("cluster_id")
  898. if err != nil {
  899. cluster = env.GetClusterID()
  900. }
  901. node, err := res.GetString("node")
  902. if err != nil {
  903. log.Warningf("CostModel.ComputeAllocation: Node GPU cost query result missing field: %s", err)
  904. continue
  905. }
  906. instanceType, err := res.GetString("instance_type")
  907. if err != nil {
  908. log.Warningf("CostModel.ComputeAllocation: Node GPU cost query result missing field: %s", err)
  909. continue
  910. }
  911. key := newNodeKey(cluster, node)
  912. if _, ok := nodeMap[key]; !ok {
  913. nodeMap[key] = &Node{
  914. Name: node,
  915. NodeType: instanceType,
  916. }
  917. }
  918. nodeMap[key].CostPerGPUHr = res.Values[0].Value
  919. }
  920. }
  921. func applyNodeSpot(nodeMap map[nodeKey]*Node, resNodeIsSpot []*prom.QueryResult) {
  922. for _, res := range resNodeIsSpot {
  923. cluster, err := res.GetString("cluster_id")
  924. if err != nil {
  925. cluster = env.GetClusterID()
  926. }
  927. node, err := res.GetString("node")
  928. if err != nil {
  929. log.Warningf("CostModel.ComputeAllocation: Node spot query result missing field: %s", err)
  930. continue
  931. }
  932. key := newNodeKey(cluster, node)
  933. if _, ok := nodeMap[key]; !ok {
  934. log.Warningf("CostModel.ComputeAllocation: Node spot query result for missing node: %s", key)
  935. continue
  936. }
  937. nodeMap[key].Preemptible = res.Values[0].Value > 0
  938. }
  939. }
  940. func applyNodeDiscount(nodeMap map[nodeKey]*Node, cm *CostModel) {
  941. if cm == nil {
  942. return
  943. }
  944. c, err := cm.Provider.GetConfig()
  945. if err != nil {
  946. log.Errorf("CostModel.ComputeAllocation: applyNodeDiscount: %s", err)
  947. return
  948. }
  949. discount, err := ParsePercentString(c.Discount)
  950. if err != nil {
  951. log.Errorf("CostModel.ComputeAllocation: applyNodeDiscount: %s", err)
  952. return
  953. }
  954. negotiatedDiscount, err := ParsePercentString(c.NegotiatedDiscount)
  955. if err != nil {
  956. log.Errorf("CostModel.ComputeAllocation: applyNodeDiscount: %s", err)
  957. return
  958. }
  959. for _, node := range nodeMap {
  960. // TODO niko/computeallocation GKE Reserved Instances into account
  961. node.Discount = cm.Provider.CombinedDiscountForNode(node.NodeType, node.Preemptible, discount, negotiatedDiscount)
  962. node.CostPerCPUHr *= (1.0 - node.Discount)
  963. node.CostPerRAMGiBHr *= (1.0 - node.Discount)
  964. }
  965. }
  966. func buildPVMap(pvMap map[pvKey]*PV, resPVCostPerGiBHour []*prom.QueryResult) {
  967. for _, res := range resPVCostPerGiBHour {
  968. cluster, err := res.GetString("cluster_id")
  969. if err != nil {
  970. cluster = env.GetClusterID()
  971. }
  972. name, err := res.GetString("volumename")
  973. if err != nil {
  974. log.Warningf("CostModel.ComputeAllocation: PV cost without volumename")
  975. continue
  976. }
  977. key := newPVKey(cluster, name)
  978. pvMap[key] = &PV{
  979. Cluster: cluster,
  980. Name: name,
  981. CostPerGiBHour: res.Values[0].Value,
  982. }
  983. }
  984. }
  985. func applyPVBytes(pvMap map[pvKey]*PV, resPVBytes []*prom.QueryResult) {
  986. for _, res := range resPVBytes {
  987. key, err := resultPVKey(res, "cluster_id", "persistentvolume")
  988. if err != nil {
  989. log.Warningf("CostModel.ComputeAllocation: PV bytes query result missing field: %s", err)
  990. continue
  991. }
  992. if _, ok := pvMap[key]; !ok {
  993. log.Warningf("CostModel.ComputeAllocation: PV bytes result for missing PV: %s", err)
  994. continue
  995. }
  996. pvMap[key].Bytes = res.Values[0].Value
  997. }
  998. }
  999. func buildPVCMap(window kubecost.Window, pvcMap map[pvcKey]*PVC, pvMap map[pvKey]*PV, resPVCInfo []*prom.QueryResult) {
  1000. for _, res := range resPVCInfo {
  1001. cluster, err := res.GetString("cluster_id")
  1002. if err != nil {
  1003. cluster = env.GetClusterID()
  1004. }
  1005. values, err := res.GetStrings("persistentvolumeclaim", "storageclass", "volumename", "namespace")
  1006. if err != nil {
  1007. log.Warningf("CostModel.ComputeAllocation: PVC info query result missing field: %s", err)
  1008. continue
  1009. }
  1010. namespace := values["namespace"]
  1011. name := values["persistentvolumeclaim"]
  1012. volume := values["volumename"]
  1013. storageClass := values["storageclass"]
  1014. pvKey := newPVKey(cluster, volume)
  1015. pvcKey := newPVCKey(cluster, namespace, name)
  1016. // pvcStart and pvcEnd are the timestamps of the first and last minutes
  1017. // the PVC was running, respectively. We subtract 1m from pvcStart
  1018. // because this point will actually represent the end of the first
  1019. // minute. We don't subtract from pvcEnd because it already represents
  1020. // the end of the last minute.
  1021. var pvcStart, pvcEnd time.Time
  1022. for _, datum := range res.Values {
  1023. t := time.Unix(int64(datum.Timestamp), 0)
  1024. if pvcStart.IsZero() && datum.Value > 0 && window.Contains(t) {
  1025. pvcStart = t
  1026. }
  1027. if datum.Value > 0 && window.Contains(t) {
  1028. pvcEnd = t
  1029. }
  1030. }
  1031. if pvcStart.IsZero() || pvcEnd.IsZero() {
  1032. log.Warningf("CostModel.ComputeAllocation: PVC %s has no running time", pvcKey)
  1033. }
  1034. pvcStart = pvcStart.Add(-time.Minute)
  1035. if _, ok := pvMap[pvKey]; !ok {
  1036. log.Warningf("CostModel.ComputeAllocation: PV missing for PVC info query result: %s", pvKey)
  1037. continue
  1038. }
  1039. pvMap[pvKey].StorageClass = storageClass
  1040. if _, ok := pvcMap[pvcKey]; !ok {
  1041. pvcMap[pvcKey] = &PVC{}
  1042. }
  1043. pvcMap[pvcKey].Name = name
  1044. pvcMap[pvcKey].Namespace = namespace
  1045. pvcMap[pvcKey].Volume = pvMap[pvKey]
  1046. pvcMap[pvcKey].Start = pvcStart
  1047. pvcMap[pvcKey].End = pvcEnd
  1048. }
  1049. }
  1050. func applyPVCBytesRequested(pvcMap map[pvcKey]*PVC, resPVCBytesRequested []*prom.QueryResult) {
  1051. for _, res := range resPVCBytesRequested {
  1052. key, err := resultPVCKey(res, "cluster_id", "namespace", "persistentvolumeclaim")
  1053. if err != nil {
  1054. log.Warningf("CostModel.ComputeAllocation: PVC bytes requested query result missing field: %s", err)
  1055. continue
  1056. }
  1057. if _, ok := pvcMap[key]; !ok {
  1058. log.Warningf("CostModel.ComputeAllocation: PVC bytes requested result for missing PVC: %s", key)
  1059. continue
  1060. }
  1061. pvcMap[key].Bytes = res.Values[0].Value
  1062. }
  1063. }
  1064. func buildPodPVCMap(podPVCMap map[podKey][]*PVC, pvMap map[pvKey]*PV, pvcMap map[pvcKey]*PVC, podAllocation map[podKey][]*kubecost.Allocation, resPodPVCAllocation []*prom.QueryResult) {
  1065. for _, res := range resPodPVCAllocation {
  1066. cluster, err := res.GetString("cluster_id")
  1067. if err != nil {
  1068. cluster = env.GetClusterID()
  1069. }
  1070. values, err := res.GetStrings("persistentvolume", "persistentvolumeclaim", "pod", "namespace")
  1071. if err != nil {
  1072. log.Warningf("CostModel.ComputeAllocation: PVC allocation query result missing field: %s", err)
  1073. continue
  1074. }
  1075. namespace := values["namespace"]
  1076. pod := values["pod"]
  1077. name := values["persistentvolumeclaim"]
  1078. volume := values["persistentvolume"]
  1079. podKey := newPodKey(cluster, namespace, pod)
  1080. pvKey := newPVKey(cluster, volume)
  1081. pvcKey := newPVCKey(cluster, namespace, name)
  1082. if _, ok := pvMap[pvKey]; !ok {
  1083. log.Warningf("CostModel.ComputeAllocation: PV missing for PVC allocation query result: %s", pvKey)
  1084. continue
  1085. }
  1086. if _, ok := podPVCMap[podKey]; !ok {
  1087. podPVCMap[podKey] = []*PVC{}
  1088. }
  1089. pvc, ok := pvcMap[pvcKey]
  1090. if !ok {
  1091. log.Warningf("CostModel.ComputeAllocation: PVC missing for PVC allocation query: %s", pvcKey)
  1092. continue
  1093. }
  1094. pvc.Count = len(podAllocation[podKey])
  1095. pvc.Mounted = true
  1096. podPVCMap[podKey] = append(podPVCMap[podKey], pvc)
  1097. }
  1098. }
  1099. func applyUnmountedPVs(window kubecost.Window, allocationMap map[containerKey]*kubecost.Allocation, pvMap map[pvKey]*PV, pvcMap map[pvcKey]*PVC) {
  1100. unmountedPVBytes := map[string]float64{}
  1101. unmountedPVCost := map[string]float64{}
  1102. for _, pv := range pvMap {
  1103. mounted := false
  1104. for _, pvc := range pvcMap {
  1105. if pvc.Volume == nil {
  1106. continue
  1107. }
  1108. if pvc.Volume == pv {
  1109. mounted = true
  1110. break
  1111. }
  1112. }
  1113. if !mounted {
  1114. gib := pv.Bytes / 1024 / 1024 / 1024
  1115. hrs := window.Minutes() / 60.0 // TODO niko/computeallocation PV hours, not window hours?
  1116. cost := pv.CostPerGiBHour * gib * hrs
  1117. unmountedPVCost[pv.Cluster] += cost
  1118. unmountedPVBytes[pv.Cluster] += pv.Bytes
  1119. }
  1120. }
  1121. for cluster, amount := range unmountedPVCost {
  1122. container := "unmounted-pvs"
  1123. pod := "unmounted-pvs"
  1124. namespace := ""
  1125. node := ""
  1126. containerKey := newContainerKey(cluster, namespace, pod, container)
  1127. allocationMap[containerKey] = &kubecost.Allocation{
  1128. Name: fmt.Sprintf("%s/%s/%s/%s/%s", cluster, node, namespace, pod, container),
  1129. Properties: kubecost.Properties{
  1130. kubecost.ClusterProp: cluster,
  1131. kubecost.NodeProp: node,
  1132. kubecost.NamespaceProp: namespace,
  1133. kubecost.PodProp: pod,
  1134. kubecost.ContainerProp: container,
  1135. },
  1136. Window: window.Clone(),
  1137. Start: *window.Start(),
  1138. End: *window.End(),
  1139. PVByteHours: unmountedPVBytes[cluster] * window.Minutes() / 60.0,
  1140. PVCost: amount,
  1141. TotalCost: amount,
  1142. }
  1143. }
  1144. }
  1145. func applyUnmountedPVCs(window kubecost.Window, allocationMap map[containerKey]*kubecost.Allocation, pvcMap map[pvcKey]*PVC) {
  1146. unmountedPVCBytes := map[namespaceKey]float64{}
  1147. unmountedPVCCost := map[namespaceKey]float64{}
  1148. for _, pvc := range pvcMap {
  1149. if !pvc.Mounted && pvc.Volume != nil {
  1150. key := newNamespaceKey(pvc.Cluster, pvc.Namespace)
  1151. gib := pvc.Volume.Bytes / 1024 / 1024 / 1024
  1152. hrs := pvc.Minutes() / 60.0
  1153. cost := pvc.Volume.CostPerGiBHour * gib * hrs
  1154. unmountedPVCCost[key] += cost
  1155. unmountedPVCBytes[key] += pvc.Volume.Bytes
  1156. }
  1157. }
  1158. for key, amount := range unmountedPVCCost {
  1159. container := "unmounted-pvs"
  1160. pod := "unmounted-pvs"
  1161. namespace := key.Namespace
  1162. node := ""
  1163. cluster := key.Cluster
  1164. containerKey := newContainerKey(cluster, namespace, pod, container)
  1165. allocationMap[containerKey] = &kubecost.Allocation{
  1166. Name: fmt.Sprintf("%s/%s/%s/%s/%s", cluster, node, namespace, pod, container),
  1167. Properties: kubecost.Properties{
  1168. kubecost.ClusterProp: cluster,
  1169. kubecost.NodeProp: node,
  1170. kubecost.NamespaceProp: namespace,
  1171. kubecost.PodProp: pod,
  1172. kubecost.ContainerProp: container,
  1173. },
  1174. Window: window.Clone(),
  1175. Start: *window.Start(),
  1176. End: *window.End(),
  1177. PVByteHours: unmountedPVCBytes[key] * window.Minutes() / 60.0,
  1178. PVCost: amount,
  1179. TotalCost: amount,
  1180. }
  1181. }
  1182. }
  1183. // PVC describes a PersistentVolumeClaim
  1184. // TODO move to pkg/kubecost? [TODO:CLEANUP]
  1185. // TODO add PersistentVolumeClaims field to type Allocation? [TODO:CLEANUP]
  1186. type PVC struct {
  1187. Bytes float64 `json:"bytes"`
  1188. Count int `json:"count"`
  1189. Name string `json:"name"`
  1190. Cluster string `json:"cluster"`
  1191. Namespace string `json:"namespace"`
  1192. Volume *PV `json:"persistentVolume"`
  1193. Mounted bool `json:"mounted"`
  1194. Start time.Time `json:"start"`
  1195. End time.Time `json:"end"`
  1196. }
  1197. // Cost computes the cumulative cost of the PVC
  1198. func (pvc *PVC) Cost() float64 {
  1199. if pvc == nil || pvc.Volume == nil {
  1200. return 0.0
  1201. }
  1202. gib := pvc.Bytes / 1024 / 1024 / 1024
  1203. hrs := pvc.Minutes() / 60.0
  1204. return pvc.Volume.CostPerGiBHour * gib * hrs
  1205. }
  1206. // Minutes computes the number of minutes over which the PVC is defined
  1207. func (pvc *PVC) Minutes() float64 {
  1208. if pvc == nil {
  1209. return 0.0
  1210. }
  1211. return pvc.End.Sub(pvc.Start).Minutes()
  1212. }
  1213. // String returns a string representation of the PVC
  1214. func (pvc *PVC) String() string {
  1215. if pvc == nil {
  1216. return "<nil>"
  1217. }
  1218. 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))
  1219. }
  1220. // PV describes a PersistentVolume
  1221. // TODO move to pkg/kubecost? [TODO:CLEANUP]
  1222. type PV struct {
  1223. Bytes float64 `json:"bytes"`
  1224. CostPerGiBHour float64 `json:"costPerGiBHour"` // TODO niko/computeallocation GiB or GB?
  1225. Cluster string `json:"cluster"`
  1226. Name string `json:"name"`
  1227. StorageClass string `json:"storageClass"`
  1228. }
  1229. // String returns a string representation of the PV
  1230. func (pv *PV) String() string {
  1231. if pv == nil {
  1232. return "<nil>"
  1233. }
  1234. return fmt.Sprintf("%s/%s{Bytes:%.2f, Cost/GiB*Hr:%.6f, StorageClass:%s}", pv.Cluster, pv.Name, pv.Bytes, pv.CostPerGiBHour, pv.StorageClass)
  1235. }