allocation.go 64 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899
  1. package costmodel
  2. import (
  3. "fmt"
  4. "math"
  5. "strconv"
  6. "strings"
  7. "time"
  8. "github.com/kubecost/cost-model/pkg/cloud"
  9. "github.com/kubecost/cost-model/pkg/env"
  10. "github.com/kubecost/cost-model/pkg/kubecost"
  11. "github.com/kubecost/cost-model/pkg/log"
  12. "github.com/kubecost/cost-model/pkg/prom"
  13. "github.com/kubecost/cost-model/pkg/util"
  14. "k8s.io/apimachinery/pkg/labels"
  15. )
  16. const (
  17. queryFmtPods = `avg(kube_pod_container_status_running{}) by (pod, namespace, cluster_id)[%s:%s]%s`
  18. queryFmtRAMBytesAllocated = `avg(avg_over_time(container_memory_allocation_bytes{container!="", container!="POD", node!=""}[%s]%s)) by (container, pod, namespace, node, cluster_id)`
  19. 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)`
  20. 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)`
  21. queryFmtCPUCoresAllocated = `avg(avg_over_time(container_cpu_allocation{container!="", container!="POD", node!=""}[%s]%s)) by (container, pod, namespace, node, cluster_id)`
  22. 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)`
  23. 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)`
  24. 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)`
  25. queryFmtNodeCostPerCPUHr = `avg(avg_over_time(node_cpu_hourly_cost[%s]%s)) by (node, cluster_id, instance_type)`
  26. queryFmtNodeCostPerRAMGiBHr = `avg(avg_over_time(node_ram_hourly_cost[%s]%s)) by (node, cluster_id, instance_type)`
  27. queryFmtNodeCostPerGPUHr = `avg(avg_over_time(node_gpu_hourly_cost[%s]%s)) by (node, cluster_id, instance_type)`
  28. queryFmtNodeIsSpot = `avg_over_time(kubecost_node_is_spot[%s]%s)`
  29. queryFmtPVCInfo = `avg(kube_persistentvolumeclaim_info{volumename != ""}) by (persistentvolumeclaim, storageclass, volumename, namespace, cluster_id)[%s:%s]%s`
  30. queryFmtPVBytes = `avg(avg_over_time(kube_persistentvolume_capacity_bytes[%s]%s)) by (persistentvolume, cluster_id)`
  31. queryFmtPodPVCAllocation = `avg(avg_over_time(pod_pvc_allocation[%s]%s)) by (persistentvolume, persistentvolumeclaim, pod, namespace, cluster_id)`
  32. queryFmtPVCBytesRequested = `avg(avg_over_time(kube_persistentvolumeclaim_resource_requests_storage_bytes{}[%s]%s)) by (persistentvolumeclaim, namespace, cluster_id)`
  33. queryFmtPVCostPerGiBHour = `avg(avg_over_time(pv_hourly_cost[%s]%s)) by (volumename, cluster_id)`
  34. 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`
  35. queryFmtNetZoneCostPerGiB = `avg(avg_over_time(kubecost_network_zone_egress_cost{}[%s]%s)) by (cluster_id)`
  36. 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`
  37. queryFmtNetRegionCostPerGiB = `avg(avg_over_time(kubecost_network_region_egress_cost{}[%s]%s)) by (cluster_id)`
  38. queryFmtNetInternetGiB = `sum(increase(kubecost_pod_network_egress_bytes_total{internet="true"}[%s]%s)) by (pod_name, namespace, cluster_id) / 1024 / 1024 / 1024`
  39. queryFmtNetInternetCostPerGiB = `avg(avg_over_time(kubecost_network_internet_egress_cost{}[%s]%s)) by (cluster_id)`
  40. queryFmtNamespaceLabels = `avg_over_time(kube_namespace_labels[%s]%s)`
  41. queryFmtNamespaceAnnotations = `avg_over_time(kube_namespace_annotations[%s]%s)`
  42. queryFmtPodLabels = `avg_over_time(kube_pod_labels[%s]%s)`
  43. queryFmtPodAnnotations = `avg_over_time(kube_pod_annotations[%s]%s)`
  44. queryFmtServiceLabels = `avg_over_time(service_selector_labels[%s]%s)`
  45. queryFmtDeploymentLabels = `avg_over_time(deployment_match_labels[%s]%s)`
  46. queryFmtStatefulSetLabels = `avg_over_time(statefulSet_match_labels[%s]%s)`
  47. queryFmtDaemonSetLabels = `sum(avg_over_time(kube_pod_owner{owner_kind="DaemonSet"}[%s]%s)) by (pod, owner_name, namespace, cluster_id)`
  48. queryFmtJobLabels = `sum(avg_over_time(kube_pod_owner{owner_kind="Job"}[%s]%s)) by (pod, owner_name, namespace ,cluster_id)`
  49. queryFmtLBCostPerHr = `avg(avg_over_time(kubecost_load_balancer_cost[%s]%s)) by (namespace, service_name, cluster_id)`
  50. queryFmtLBActiveMins = `count(kubecost_load_balancer_cost) by (namespace, service_name, cluster_id)[%s:%s]%s`
  51. )
  52. // ComputeAllocation uses the CostModel instance to compute an AllocationSet
  53. // for the window defined by the given start and end times. The Allocations
  54. // returned are unaggregated (i.e. down to the container level).
  55. func (cm *CostModel) ComputeAllocation(start, end time.Time, resolution time.Duration) (*kubecost.AllocationSet, error) {
  56. // 1. Build out Pod map from resolution-tuned, batched Pod start/end query
  57. // 2. Run and apply the results of the remaining queries to
  58. // 3. Build out AllocationSet from completed Pod map
  59. // Create a window spanning the requested query
  60. window := kubecost.NewWindow(&start, &end)
  61. // Create an empty AllocationSet. For safety, in the case of an error, we
  62. // should prefer to return this empty set with the error. (In the case of
  63. // no error, of course we populate the set and return it.)
  64. allocSet := kubecost.NewAllocationSet(start, end)
  65. // (1) Build out Pod map
  66. // Build out a map of Allocations as a mapping from pod-to-container-to-
  67. // underlying-Allocation instance, starting with (start, end) so that we
  68. // begin with minutes, from which we compute resource allocation and cost
  69. // totals from measured rate data.
  70. podMap := map[podKey]*Pod{}
  71. // clusterStarts and clusterEnds record the earliest start and latest end
  72. // times, respectively, on a cluster-basis. These are used for unmounted
  73. // PVs and other "virtual" Allocations so that minutes are maximally
  74. // accurate during start-up or spin-down of a cluster
  75. clusterStart := map[string]time.Time{}
  76. clusterEnd := map[string]time.Time{}
  77. cm.buildPodMap(window, resolution, env.GetETLMaxBatchDuration(), podMap, clusterStart, clusterEnd)
  78. // (2) Run and apply remaining queries
  79. // Convert window (start, end) to (duration, offset) for querying Prometheus,
  80. // including handling Thanos offset
  81. durStr, offStr, err := window.DurationOffsetForPrometheus()
  82. if err != nil {
  83. // Negative duration, so return empty set
  84. return allocSet, nil
  85. }
  86. // Convert resolution duration to a query-ready string
  87. resStr := util.DurationString(resolution)
  88. ctx := prom.NewContext(cm.PrometheusClient)
  89. queryRAMBytesAllocated := fmt.Sprintf(queryFmtRAMBytesAllocated, durStr, offStr)
  90. resChRAMBytesAllocated := ctx.Query(queryRAMBytesAllocated)
  91. queryRAMRequests := fmt.Sprintf(queryFmtRAMRequests, durStr, offStr)
  92. resChRAMRequests := ctx.Query(queryRAMRequests)
  93. queryRAMUsage := fmt.Sprintf(queryFmtRAMUsage, durStr, offStr)
  94. resChRAMUsage := ctx.Query(queryRAMUsage)
  95. queryCPUCoresAllocated := fmt.Sprintf(queryFmtCPUCoresAllocated, durStr, offStr)
  96. resChCPUCoresAllocated := ctx.Query(queryCPUCoresAllocated)
  97. queryCPURequests := fmt.Sprintf(queryFmtCPURequests, durStr, offStr)
  98. resChCPURequests := ctx.Query(queryCPURequests)
  99. queryCPUUsage := fmt.Sprintf(queryFmtCPUUsage, durStr, offStr)
  100. resChCPUUsage := ctx.Query(queryCPUUsage)
  101. queryGPUsRequested := fmt.Sprintf(queryFmtGPUsRequested, durStr, offStr)
  102. resChGPUsRequested := ctx.Query(queryGPUsRequested)
  103. queryNodeCostPerCPUHr := fmt.Sprintf(queryFmtNodeCostPerCPUHr, durStr, offStr)
  104. resChNodeCostPerCPUHr := ctx.Query(queryNodeCostPerCPUHr)
  105. queryNodeCostPerRAMGiBHr := fmt.Sprintf(queryFmtNodeCostPerRAMGiBHr, durStr, offStr)
  106. resChNodeCostPerRAMGiBHr := ctx.Query(queryNodeCostPerRAMGiBHr)
  107. queryNodeCostPerGPUHr := fmt.Sprintf(queryFmtNodeCostPerGPUHr, durStr, offStr)
  108. resChNodeCostPerGPUHr := ctx.Query(queryNodeCostPerGPUHr)
  109. queryNodeIsSpot := fmt.Sprintf(queryFmtNodeIsSpot, durStr, offStr)
  110. resChNodeIsSpot := ctx.Query(queryNodeIsSpot)
  111. queryPVCInfo := fmt.Sprintf(queryFmtPVCInfo, durStr, resStr, offStr)
  112. resChPVCInfo := ctx.Query(queryPVCInfo)
  113. queryPVBytes := fmt.Sprintf(queryFmtPVBytes, durStr, offStr)
  114. resChPVBytes := ctx.Query(queryPVBytes)
  115. queryPodPVCAllocation := fmt.Sprintf(queryFmtPodPVCAllocation, durStr, offStr)
  116. resChPodPVCAllocation := ctx.Query(queryPodPVCAllocation)
  117. queryPVCBytesRequested := fmt.Sprintf(queryFmtPVCBytesRequested, durStr, offStr)
  118. resChPVCBytesRequested := ctx.Query(queryPVCBytesRequested)
  119. queryPVCostPerGiBHour := fmt.Sprintf(queryFmtPVCostPerGiBHour, durStr, offStr)
  120. resChPVCostPerGiBHour := ctx.Query(queryPVCostPerGiBHour)
  121. queryNetZoneGiB := fmt.Sprintf(queryFmtNetZoneGiB, durStr, offStr)
  122. resChNetZoneGiB := ctx.Query(queryNetZoneGiB)
  123. queryNetZoneCostPerGiB := fmt.Sprintf(queryFmtNetZoneCostPerGiB, durStr, offStr)
  124. resChNetZoneCostPerGiB := ctx.Query(queryNetZoneCostPerGiB)
  125. queryNetRegionGiB := fmt.Sprintf(queryFmtNetRegionGiB, durStr, offStr)
  126. resChNetRegionGiB := ctx.Query(queryNetRegionGiB)
  127. queryNetRegionCostPerGiB := fmt.Sprintf(queryFmtNetRegionCostPerGiB, durStr, offStr)
  128. resChNetRegionCostPerGiB := ctx.Query(queryNetRegionCostPerGiB)
  129. queryNetInternetGiB := fmt.Sprintf(queryFmtNetInternetGiB, durStr, offStr)
  130. resChNetInternetGiB := ctx.Query(queryNetInternetGiB)
  131. queryNetInternetCostPerGiB := fmt.Sprintf(queryFmtNetInternetCostPerGiB, durStr, offStr)
  132. resChNetInternetCostPerGiB := ctx.Query(queryNetInternetCostPerGiB)
  133. queryNamespaceLabels := fmt.Sprintf(queryFmtNamespaceLabels, durStr, offStr)
  134. resChNamespaceLabels := ctx.Query(queryNamespaceLabels)
  135. queryNamespaceAnnotations := fmt.Sprintf(queryFmtNamespaceAnnotations, durStr, offStr)
  136. resChNamespaceAnnotations := ctx.Query(queryNamespaceAnnotations)
  137. queryPodLabels := fmt.Sprintf(queryFmtPodLabels, durStr, offStr)
  138. resChPodLabels := ctx.Query(queryPodLabels)
  139. queryPodAnnotations := fmt.Sprintf(queryFmtPodAnnotations, durStr, offStr)
  140. resChPodAnnotations := ctx.Query(queryPodAnnotations)
  141. queryServiceLabels := fmt.Sprintf(queryFmtServiceLabels, durStr, offStr)
  142. resChServiceLabels := ctx.Query(queryServiceLabels)
  143. queryDeploymentLabels := fmt.Sprintf(queryFmtDeploymentLabels, durStr, offStr)
  144. resChDeploymentLabels := ctx.Query(queryDeploymentLabels)
  145. queryStatefulSetLabels := fmt.Sprintf(queryFmtStatefulSetLabels, durStr, offStr)
  146. resChStatefulSetLabels := ctx.Query(queryStatefulSetLabels)
  147. queryDaemonSetLabels := fmt.Sprintf(queryFmtDaemonSetLabels, durStr, offStr)
  148. resChDaemonSetLabels := ctx.Query(queryDaemonSetLabels)
  149. queryJobLabels := fmt.Sprintf(queryFmtJobLabels, durStr, offStr)
  150. resChJobLabels := ctx.Query(queryJobLabels)
  151. queryLBCostPerHr := fmt.Sprintf(queryFmtLBCostPerHr, durStr, offStr)
  152. resChLBCostPerHr := ctx.Query(queryLBCostPerHr)
  153. queryLBActiveMins := fmt.Sprintf(queryFmtLBActiveMins, durStr, resStr, offStr)
  154. resChLBActiveMins := ctx.Query(queryLBActiveMins)
  155. resCPUCoresAllocated, _ := resChCPUCoresAllocated.Await()
  156. resCPURequests, _ := resChCPURequests.Await()
  157. resCPUUsage, _ := resChCPUUsage.Await()
  158. resRAMBytesAllocated, _ := resChRAMBytesAllocated.Await()
  159. resRAMRequests, _ := resChRAMRequests.Await()
  160. resRAMUsage, _ := resChRAMUsage.Await()
  161. resGPUsRequested, _ := resChGPUsRequested.Await()
  162. resNodeCostPerCPUHr, _ := resChNodeCostPerCPUHr.Await()
  163. resNodeCostPerRAMGiBHr, _ := resChNodeCostPerRAMGiBHr.Await()
  164. resNodeCostPerGPUHr, _ := resChNodeCostPerGPUHr.Await()
  165. resNodeIsSpot, _ := resChNodeIsSpot.Await()
  166. resPVBytes, _ := resChPVBytes.Await()
  167. resPVCostPerGiBHour, _ := resChPVCostPerGiBHour.Await()
  168. resPVCInfo, _ := resChPVCInfo.Await()
  169. resPVCBytesRequested, _ := resChPVCBytesRequested.Await()
  170. resPodPVCAllocation, _ := resChPodPVCAllocation.Await()
  171. resNetZoneGiB, _ := resChNetZoneGiB.Await()
  172. resNetZoneCostPerGiB, _ := resChNetZoneCostPerGiB.Await()
  173. resNetRegionGiB, _ := resChNetRegionGiB.Await()
  174. resNetRegionCostPerGiB, _ := resChNetRegionCostPerGiB.Await()
  175. resNetInternetGiB, _ := resChNetInternetGiB.Await()
  176. resNetInternetCostPerGiB, _ := resChNetInternetCostPerGiB.Await()
  177. resNamespaceLabels, _ := resChNamespaceLabels.Await()
  178. resNamespaceAnnotations, _ := resChNamespaceAnnotations.Await()
  179. resPodLabels, _ := resChPodLabels.Await()
  180. resPodAnnotations, _ := resChPodAnnotations.Await()
  181. resServiceLabels, _ := resChServiceLabels.Await()
  182. resDeploymentLabels, _ := resChDeploymentLabels.Await()
  183. resStatefulSetLabels, _ := resChStatefulSetLabels.Await()
  184. resDaemonSetLabels, _ := resChDaemonSetLabels.Await()
  185. resJobLabels, _ := resChJobLabels.Await()
  186. resLBCostPerHr, _ := resChLBCostPerHr.Await()
  187. resLBActiveMins, _ := resChLBActiveMins.Await()
  188. if ctx.HasErrors() {
  189. for _, err := range ctx.Errors() {
  190. log.Errorf("CostModel.ComputeAllocation: %s", err)
  191. }
  192. return allocSet, ctx.ErrorCollection()
  193. }
  194. // We choose to apply allocation before requests in the cases of RAM and
  195. // CPU so that we can assert that allocation should always be greater than
  196. // or equal to request.
  197. applyCPUCoresAllocated(podMap, resCPUCoresAllocated)
  198. applyCPUCoresRequested(podMap, resCPURequests)
  199. applyCPUCoresUsed(podMap, resCPUUsage)
  200. applyRAMBytesAllocated(podMap, resRAMBytesAllocated)
  201. applyRAMBytesRequested(podMap, resRAMRequests)
  202. applyRAMBytesUsed(podMap, resRAMUsage)
  203. applyGPUsRequested(podMap, resGPUsRequested)
  204. applyNetworkAllocation(podMap, resNetZoneGiB, resNetZoneCostPerGiB)
  205. applyNetworkAllocation(podMap, resNetRegionGiB, resNetRegionCostPerGiB)
  206. applyNetworkAllocation(podMap, resNetInternetGiB, resNetInternetCostPerGiB)
  207. namespaceLabels := resToNamespaceLabels(resNamespaceLabels)
  208. podLabels := resToPodLabels(resPodLabels)
  209. namespaceAnnotations := resToNamespaceAnnotations(resNamespaceAnnotations)
  210. podAnnotations := resToPodAnnotations(resPodAnnotations)
  211. applyLabels(podMap, namespaceLabels, podLabels)
  212. applyAnnotations(podMap, namespaceAnnotations, podAnnotations)
  213. serviceLabels := getServiceLabels(resServiceLabels)
  214. allocsByService := map[serviceKey][]*kubecost.Allocation{}
  215. applyServicesToPods(podMap, podLabels, allocsByService, serviceLabels)
  216. podDeploymentMap := labelsToPodControllerMap(podLabels, resToDeploymentLabels(resDeploymentLabels))
  217. podStatefulSetMap := labelsToPodControllerMap(podLabels, resToStatefulSetLabels(resStatefulSetLabels))
  218. podDaemonSetMap := resToPodDaemonSetMap(resDaemonSetLabels)
  219. podJobMap := resToPodJobMap(resJobLabels)
  220. applyControllersToPods(podMap, podDeploymentMap)
  221. applyControllersToPods(podMap, podStatefulSetMap)
  222. applyControllersToPods(podMap, podDaemonSetMap)
  223. applyControllersToPods(podMap, podJobMap)
  224. // TODO breakdown network costs?
  225. // Build out a map of Nodes with resource costs, discounts, and node types
  226. // for converting resource allocation data to cumulative costs.
  227. nodeMap := map[nodeKey]*NodePricing{}
  228. applyNodeCostPerCPUHr(nodeMap, resNodeCostPerCPUHr)
  229. applyNodeCostPerRAMGiBHr(nodeMap, resNodeCostPerRAMGiBHr)
  230. applyNodeCostPerGPUHr(nodeMap, resNodeCostPerGPUHr)
  231. applyNodeSpot(nodeMap, resNodeIsSpot)
  232. applyNodeDiscount(nodeMap, cm)
  233. // Build out the map of all PVs with class, size and cost-per-hour.
  234. // Note: this does not record time running, which we may want to
  235. // include later for increased PV precision. (As long as the PV has
  236. // a PVC, we get time running there, so this is only inaccurate
  237. // for short-lived, unmounted PVs.)
  238. pvMap := map[pvKey]*PV{}
  239. buildPVMap(pvMap, resPVCostPerGiBHour)
  240. applyPVBytes(pvMap, resPVBytes)
  241. // Build out the map of all PVCs with time running, bytes requested,
  242. // and connect to the correct PV from pvMap. (If no PV exists, that
  243. // is noted, but does not result in any allocation/cost.)
  244. pvcMap := map[pvcKey]*PVC{}
  245. buildPVCMap(window, pvcMap, pvMap, resPVCInfo)
  246. applyPVCBytesRequested(pvcMap, resPVCBytesRequested)
  247. // Build out the relationships of pods to their PVCs. This step
  248. // populates the PVC.Count field so that PVC allocation can be
  249. // split appropriately among each pod's container allocation.
  250. podPVCMap := map[podKey][]*PVC{}
  251. buildPodPVCMap(podPVCMap, pvMap, pvcMap, podMap, resPodPVCAllocation)
  252. // Identify unmounted PVs (PVs without PVCs) and add one Allocation per
  253. // cluster representing each cluster's unmounted PVs (if necessary).
  254. applyUnmountedPVs(window, podMap, pvMap, pvcMap)
  255. lbMap := getLoadBalancerCosts(resLBCostPerHr, resLBActiveMins, resolution)
  256. applyLoadBalancersToPods(lbMap, allocsByService)
  257. // (3) Build out AllocationSet from Pod map
  258. for _, pod := range podMap {
  259. for _, alloc := range pod.Allocations {
  260. cluster, _ := alloc.Properties.GetCluster()
  261. nodeName, _ := alloc.Properties.GetNode()
  262. namespace, _ := alloc.Properties.GetNamespace()
  263. pod, _ := alloc.Properties.GetPod()
  264. container, _ := alloc.Properties.GetContainer()
  265. podKey := newPodKey(cluster, namespace, pod)
  266. nodeKey := newNodeKey(cluster, nodeName)
  267. node := cm.getNodePricing(nodeMap, nodeKey)
  268. alloc.CPUCost = alloc.CPUCoreHours * node.CostPerCPUHr
  269. alloc.RAMCost = (alloc.RAMByteHours / 1024 / 1024 / 1024) * node.CostPerRAMGiBHr
  270. alloc.GPUCost = alloc.GPUHours * node.CostPerGPUHr
  271. if pvcs, ok := podPVCMap[podKey]; ok {
  272. for _, pvc := range pvcs {
  273. // Determine the (start, end) of the relationship between the
  274. // given PVC and the associated Allocation so that a precise
  275. // number of hours can be used to compute cumulative cost.
  276. s, e := alloc.Start, alloc.End
  277. if pvc.Start.After(alloc.Start) {
  278. s = pvc.Start
  279. }
  280. if pvc.End.Before(alloc.End) {
  281. e = pvc.End
  282. }
  283. minutes := e.Sub(s).Minutes()
  284. hrs := minutes / 60.0
  285. count := float64(pvc.Count)
  286. if pvc.Count < 1 {
  287. count = 1
  288. }
  289. gib := pvc.Bytes / 1024 / 1024 / 1024
  290. cost := pvc.Volume.CostPerGiBHour * gib * hrs
  291. // Apply the size and cost of the PV to the allocation, each
  292. // weighted by count (i.e. the number of containers in the pod)
  293. alloc.PVByteHours += pvc.Bytes * hrs / count
  294. alloc.PVCost += cost / count
  295. }
  296. }
  297. // Make sure that the name is correct (node may not be present at this
  298. // point due to it missing from queryMinutes) then insert.
  299. alloc.Name = fmt.Sprintf("%s/%s/%s/%s/%s", cluster, nodeName, namespace, pod, container)
  300. allocSet.Set(alloc)
  301. }
  302. }
  303. return allocSet, nil
  304. }
  305. func (cm *CostModel) buildPodMap(window kubecost.Window, resolution, maxBatchSize time.Duration, podMap map[podKey]*Pod, clusterStart, clusterEnd map[string]time.Time) error {
  306. // Assumes that window is positive and closed
  307. start, end := *window.Start(), *window.End()
  308. // Convert resolution duration to a query-ready string
  309. resStr := util.DurationString(resolution)
  310. ctx := prom.NewContext(cm.PrometheusClient)
  311. // Query for (start, end) by (pod, namespace, cluster) over the given
  312. // window, using the given resolution, and if necessary in batches no
  313. // larger than the given maximum batch size. If working in batches, track
  314. // overall progress by starting with (window.start, window.start) and
  315. // querying in batches no larger than maxBatchSize from start-to-end,
  316. // folding each result set into podMap as the results come back.
  317. coverage := kubecost.NewWindow(&start, &start)
  318. numQuery := 1
  319. for coverage.End().Before(end) {
  320. // Determine the (start, end) of the current batch
  321. batchStart := *coverage.End()
  322. batchEnd := coverage.End().Add(maxBatchSize)
  323. if batchEnd.After(end) {
  324. batchEnd = end
  325. }
  326. batchWindow := kubecost.NewWindow(&batchStart, &batchEnd)
  327. var resPods []*prom.QueryResult
  328. var err error
  329. maxTries := 3
  330. numTries := 0
  331. for resPods == nil && numTries < maxTries {
  332. numTries++
  333. // Convert window (start, end) to (duration, offset) for querying Prometheus,
  334. // including handling Thanos offset
  335. durStr, offStr, err := batchWindow.DurationOffsetForPrometheus()
  336. if err != nil || durStr == "" {
  337. // Negative duration, so set empty results and don't query
  338. resPods = []*prom.QueryResult{}
  339. err = nil
  340. break
  341. }
  342. // Submit and profile query
  343. queryPods := fmt.Sprintf(queryFmtPods, durStr, resStr, offStr)
  344. queryProfile := time.Now()
  345. resPods, err = ctx.Query(queryPods).Await()
  346. if err != nil {
  347. log.Profile(queryProfile, fmt.Sprintf("CostModel.ComputeAllocation: pod query %d try %d failed: %s", numQuery, numTries, queryPods))
  348. resPods = nil
  349. }
  350. }
  351. if err != nil {
  352. return err
  353. }
  354. applyPodResults(window, resolution, podMap, clusterStart, clusterEnd, resPods)
  355. coverage = coverage.ExpandEnd(batchEnd)
  356. numQuery++
  357. }
  358. return nil
  359. }
  360. func applyPodResults(window kubecost.Window, resolution time.Duration, podMap map[podKey]*Pod, clusterStart, clusterEnd map[string]time.Time, resPods []*prom.QueryResult) {
  361. for _, res := range resPods {
  362. if len(res.Values) == 0 {
  363. log.Warningf("CostModel.ComputeAllocation: empty minutes result")
  364. continue
  365. }
  366. cluster, err := res.GetString("cluster_id")
  367. if err != nil {
  368. cluster = env.GetClusterID()
  369. }
  370. labels, err := res.GetStrings("namespace", "pod")
  371. if err != nil {
  372. log.Warningf("CostModel.ComputeAllocation: minutes query result missing field: %s", err)
  373. continue
  374. }
  375. namespace := labels["namespace"]
  376. pod := labels["pod"]
  377. key := newPodKey(cluster, namespace, pod)
  378. // allocStart and allocEnd are the timestamps of the first and last
  379. // minutes the pod was running, respectively. We subtract one resolution
  380. // from allocStart because this point will actually represent the end
  381. // of the first minute. We don't subtract from allocEnd because it
  382. // already represents the end of the last minute.
  383. var allocStart, allocEnd time.Time
  384. startAdjustmentCoeff, endAdjustmentCoeff := 1.0, 1.0
  385. for _, datum := range res.Values {
  386. t := time.Unix(int64(datum.Timestamp), 0)
  387. if allocStart.IsZero() && datum.Value > 0 && window.Contains(t) {
  388. // Set the start timestamp to the earliest non-zero timestamp
  389. allocStart = t
  390. // Record adjustment coefficient, i.e. the portion of the start
  391. // timestamp to "ignore". That is, sometimes the value will be
  392. // 0.5, meaning that we should discount the time running by
  393. // half of the resolution the timestamp stands for.
  394. startAdjustmentCoeff = (1.0 - datum.Value)
  395. }
  396. if datum.Value > 0 && window.Contains(t) {
  397. // Set the end timestamp to the latest non-zero timestamp
  398. allocEnd = t
  399. // Record adjustment coefficient, i.e. the portion of the end
  400. // timestamp to "ignore". (See explanation above for start.)
  401. endAdjustmentCoeff = (1.0 - datum.Value)
  402. }
  403. }
  404. if allocStart.IsZero() || allocEnd.IsZero() {
  405. continue
  406. }
  407. // Adjust timestamps according to the resolution and the adjustment
  408. // coefficients, as described above. That is, count the start timestamp
  409. // from the beginning of the resolution, not the end. Then "reduce" the
  410. // start and end by the correct amount, in the case that the "running"
  411. // value of the first or last timestamp was not a full 1.0.
  412. allocStart = allocStart.Add(-resolution)
  413. // Note: the *100 and /100 are necessary because Duration is an int, so
  414. // 0.5, for instance, will be truncated, resulting in no adjustment.
  415. allocStart = allocStart.Add(time.Duration(startAdjustmentCoeff*100) * resolution / time.Duration(100))
  416. allocEnd = allocEnd.Add(-time.Duration(endAdjustmentCoeff*100) * resolution / time.Duration(100))
  417. // If there is only one point with a value <= 0.5 that the start and
  418. // end timestamps both share, then we will enter this case because at
  419. // least half of a resolution will be subtracted from both the start
  420. // and the end. If that is the case, then add back half of each side
  421. // so that the pod is said to run for half a resolution total.
  422. // e.g. For resolution 1m and a value of 0.5 at one timestamp, we'll
  423. // end up with allocEnd == allocStart and each coeff == 0.5. In
  424. // that case, add 0.25m to each side, resulting in 0.5m duration.
  425. if !allocEnd.After(allocStart) {
  426. allocStart = allocStart.Add(-time.Duration(50*startAdjustmentCoeff) * resolution / time.Duration(100))
  427. allocEnd = allocEnd.Add(time.Duration(50*endAdjustmentCoeff) * resolution / time.Duration(100))
  428. }
  429. // Set start if unset or this datum's start time is earlier than the
  430. // current earliest time.
  431. if _, ok := clusterStart[cluster]; !ok || allocStart.Before(clusterStart[cluster]) {
  432. clusterStart[cluster] = allocStart
  433. }
  434. // Set end if unset or this datum's end time is later than the
  435. // current latest time.
  436. if _, ok := clusterEnd[cluster]; !ok || allocEnd.After(clusterEnd[cluster]) {
  437. clusterEnd[cluster] = allocEnd
  438. }
  439. if pod, ok := podMap[key]; ok {
  440. // Pod has already been recorded, so update it accordingly
  441. if allocStart.Before(pod.Start) {
  442. pod.Start = allocStart
  443. }
  444. if allocEnd.After(pod.End) {
  445. pod.End = allocEnd
  446. }
  447. } else {
  448. // Pod has not been recorded yet, so insert it
  449. podMap[key] = &Pod{
  450. Window: window.Clone(),
  451. Start: allocStart,
  452. End: allocEnd,
  453. Key: key,
  454. Allocations: map[string]*kubecost.Allocation{},
  455. }
  456. }
  457. }
  458. }
  459. func applyCPUCoresAllocated(podMap map[podKey]*Pod, resCPUCoresAllocated []*prom.QueryResult) {
  460. for _, res := range resCPUCoresAllocated {
  461. key, err := resultPodKey(res, "cluster_id", "namespace", "pod")
  462. if err != nil {
  463. log.DedupedWarningf(10, "CostModel.ComputeAllocation: CPU allocation result missing field: %s", err)
  464. continue
  465. }
  466. pod, ok := podMap[key]
  467. if !ok {
  468. continue
  469. }
  470. container, err := res.GetString("container")
  471. if err != nil {
  472. log.DedupedWarningf(10, "CostModel.ComputeAllocation: CPU allocation query result missing 'container': %s", key)
  473. continue
  474. }
  475. if _, ok := pod.Allocations[container]; !ok {
  476. pod.AppendContainer(container)
  477. }
  478. cpuCores := res.Values[0].Value
  479. hours := pod.Allocations[container].Minutes() / 60.0
  480. pod.Allocations[container].CPUCoreHours = cpuCores * hours
  481. node, err := res.GetString("node")
  482. if err != nil {
  483. log.Warningf("CostModel.ComputeAllocation: CPU allocation query result missing 'node': %s", key)
  484. continue
  485. }
  486. pod.Allocations[container].Properties.SetNode(node)
  487. }
  488. }
  489. func applyCPUCoresRequested(podMap map[podKey]*Pod, resCPUCoresRequested []*prom.QueryResult) {
  490. for _, res := range resCPUCoresRequested {
  491. key, err := resultPodKey(res, "cluster_id", "namespace", "pod")
  492. if err != nil {
  493. log.DedupedWarningf(10, "CostModel.ComputeAllocation: CPU request result missing field: %s", err)
  494. continue
  495. }
  496. pod, ok := podMap[key]
  497. if !ok {
  498. continue
  499. }
  500. container, err := res.GetString("container")
  501. if err != nil {
  502. log.DedupedWarningf(10, "CostModel.ComputeAllocation: CPU request query result missing 'container': %s", key)
  503. continue
  504. }
  505. if _, ok := pod.Allocations[container]; !ok {
  506. pod.AppendContainer(container)
  507. }
  508. pod.Allocations[container].CPUCoreRequestAverage = res.Values[0].Value
  509. // If CPU allocation is less than requests, set CPUCoreHours to
  510. // request level.
  511. if pod.Allocations[container].CPUCores() < res.Values[0].Value {
  512. pod.Allocations[container].CPUCoreHours = res.Values[0].Value * (pod.Allocations[container].Minutes() / 60.0)
  513. }
  514. node, err := res.GetString("node")
  515. if err != nil {
  516. log.Warningf("CostModel.ComputeAllocation: CPU request query result missing 'node': %s", key)
  517. continue
  518. }
  519. pod.Allocations[container].Properties.SetNode(node)
  520. }
  521. }
  522. func applyCPUCoresUsed(podMap map[podKey]*Pod, resCPUCoresUsed []*prom.QueryResult) {
  523. for _, res := range resCPUCoresUsed {
  524. key, err := resultPodKey(res, "cluster_id", "namespace", "pod_name")
  525. if err != nil {
  526. log.DedupedWarningf(10, "CostModel.ComputeAllocation: CPU usage result missing field: %s", err)
  527. continue
  528. }
  529. pod, ok := podMap[key]
  530. if !ok {
  531. continue
  532. }
  533. container, err := res.GetString("container_name")
  534. if err != nil {
  535. log.DedupedWarningf(10, "CostModel.ComputeAllocation: CPU usage query result missing 'container': %s", key)
  536. continue
  537. }
  538. if _, ok := pod.Allocations[container]; !ok {
  539. pod.AppendContainer(container)
  540. }
  541. pod.Allocations[container].CPUCoreUsageAverage = res.Values[0].Value
  542. }
  543. }
  544. func applyRAMBytesAllocated(podMap map[podKey]*Pod, resRAMBytesAllocated []*prom.QueryResult) {
  545. for _, res := range resRAMBytesAllocated {
  546. key, err := resultPodKey(res, "cluster_id", "namespace", "pod")
  547. if err != nil {
  548. log.DedupedWarningf(10, "CostModel.ComputeAllocation: RAM allocation result missing field: %s", err)
  549. continue
  550. }
  551. pod, ok := podMap[key]
  552. if !ok {
  553. continue
  554. }
  555. container, err := res.GetString("container")
  556. if err != nil {
  557. log.DedupedWarningf(10, "CostModel.ComputeAllocation: RAM allocation query result missing 'container': %s", key)
  558. continue
  559. }
  560. if _, ok := pod.Allocations[container]; !ok {
  561. pod.AppendContainer(container)
  562. }
  563. ramBytes := res.Values[0].Value
  564. hours := pod.Allocations[container].Minutes() / 60.0
  565. pod.Allocations[container].RAMByteHours = ramBytes * hours
  566. node, err := res.GetString("node")
  567. if err != nil {
  568. log.Warningf("CostModel.ComputeAllocation: RAM allocation query result missing 'node': %s", key)
  569. continue
  570. }
  571. pod.Allocations[container].Properties.SetNode(node)
  572. }
  573. }
  574. func applyRAMBytesRequested(podMap map[podKey]*Pod, resRAMBytesRequested []*prom.QueryResult) {
  575. for _, res := range resRAMBytesRequested {
  576. key, err := resultPodKey(res, "cluster_id", "namespace", "pod")
  577. if err != nil {
  578. log.DedupedWarningf(10, "CostModel.ComputeAllocation: RAM request result missing field: %s", err)
  579. continue
  580. }
  581. pod, ok := podMap[key]
  582. if !ok {
  583. continue
  584. }
  585. container, err := res.GetString("container")
  586. if err != nil {
  587. log.DedupedWarningf(10, "CostModel.ComputeAllocation: RAM request query result missing 'container': %s", key)
  588. continue
  589. }
  590. if _, ok := pod.Allocations[container]; !ok {
  591. pod.AppendContainer(container)
  592. }
  593. pod.Allocations[container].RAMBytesRequestAverage = res.Values[0].Value
  594. // If RAM allocation is less than requests, set RAMByteHours to
  595. // request level.
  596. if pod.Allocations[container].RAMBytes() < res.Values[0].Value {
  597. pod.Allocations[container].RAMByteHours = res.Values[0].Value * (pod.Allocations[container].Minutes() / 60.0)
  598. }
  599. node, err := res.GetString("node")
  600. if err != nil {
  601. log.Warningf("CostModel.ComputeAllocation: RAM request query result missing 'node': %s", key)
  602. continue
  603. }
  604. pod.Allocations[container].Properties.SetNode(node)
  605. }
  606. }
  607. func applyRAMBytesUsed(podMap map[podKey]*Pod, resRAMBytesUsed []*prom.QueryResult) {
  608. for _, res := range resRAMBytesUsed {
  609. key, err := resultPodKey(res, "cluster_id", "namespace", "pod_name")
  610. if err != nil {
  611. log.DedupedWarningf(10, "CostModel.ComputeAllocation: RAM usage result missing field: %s", err)
  612. continue
  613. }
  614. pod, ok := podMap[key]
  615. if !ok {
  616. continue
  617. }
  618. container, err := res.GetString("container_name")
  619. if err != nil {
  620. log.DedupedWarningf(10, "CostModel.ComputeAllocation: RAM usage query result missing 'container': %s", key)
  621. continue
  622. }
  623. if _, ok := pod.Allocations[container]; !ok {
  624. pod.AppendContainer(container)
  625. }
  626. pod.Allocations[container].RAMBytesUsageAverage = res.Values[0].Value
  627. }
  628. }
  629. func applyGPUsRequested(podMap map[podKey]*Pod, resGPUsRequested []*prom.QueryResult) {
  630. for _, res := range resGPUsRequested {
  631. key, err := resultPodKey(res, "cluster_id", "namespace", "pod")
  632. if err != nil {
  633. log.DedupedWarningf(10, "CostModel.ComputeAllocation: GPU request result missing field: %s", err)
  634. continue
  635. }
  636. pod, ok := podMap[key]
  637. if !ok {
  638. continue
  639. }
  640. container, err := res.GetString("container")
  641. if err != nil {
  642. log.DedupedWarningf(10, "CostModel.ComputeAllocation: GPU request query result missing 'container': %s", key)
  643. continue
  644. }
  645. if _, ok := pod.Allocations[container]; !ok {
  646. pod.AppendContainer(container)
  647. }
  648. hrs := pod.Allocations[container].Minutes() / 60.0
  649. pod.Allocations[container].GPUHours = res.Values[0].Value * hrs
  650. }
  651. }
  652. func applyNetworkAllocation(podMap map[podKey]*Pod, resNetworkGiB []*prom.QueryResult, resNetworkCostPerGiB []*prom.QueryResult) {
  653. costPerGiBByCluster := map[string]float64{}
  654. for _, res := range resNetworkCostPerGiB {
  655. cluster, err := res.GetString("cluster_id")
  656. if err != nil {
  657. cluster = env.GetClusterID()
  658. }
  659. costPerGiBByCluster[cluster] = res.Values[0].Value
  660. }
  661. for _, res := range resNetworkGiB {
  662. podKey, err := resultPodKey(res, "cluster_id", "namespace", "pod_name")
  663. if err != nil {
  664. log.DedupedWarningf(10, "CostModel.ComputeAllocation: Network allocation query result missing field: %s", err)
  665. continue
  666. }
  667. pod, ok := podMap[podKey]
  668. if !ok {
  669. continue
  670. }
  671. for _, alloc := range pod.Allocations {
  672. gib := res.Values[0].Value / float64(len(pod.Allocations))
  673. costPerGiB := costPerGiBByCluster[podKey.Cluster]
  674. alloc.NetworkCost = gib * costPerGiB
  675. }
  676. }
  677. }
  678. func resToNamespaceLabels(resNamespaceLabels []*prom.QueryResult) map[string]map[string]string {
  679. namespaceLabels := map[string]map[string]string{}
  680. for _, res := range resNamespaceLabels {
  681. namespace, err := res.GetString("namespace")
  682. if err != nil {
  683. continue
  684. }
  685. if _, ok := namespaceLabels[namespace]; !ok {
  686. namespaceLabels[namespace] = map[string]string{}
  687. }
  688. for k, l := range res.GetLabels() {
  689. namespaceLabels[namespace][k] = l
  690. }
  691. }
  692. return namespaceLabels
  693. }
  694. func resToPodLabels(resPodLabels []*prom.QueryResult) map[podKey]map[string]string {
  695. podLabels := map[podKey]map[string]string{}
  696. for _, res := range resPodLabels {
  697. podKey, err := resultPodKey(res, "cluster_id", "namespace", "pod")
  698. if err != nil {
  699. continue
  700. }
  701. if _, ok := podLabels[podKey]; !ok {
  702. podLabels[podKey] = map[string]string{}
  703. }
  704. for k, l := range res.GetLabels() {
  705. podLabels[podKey][k] = l
  706. }
  707. }
  708. return podLabels
  709. }
  710. func resToNamespaceAnnotations(resNamespaceAnnotations []*prom.QueryResult) map[string]map[string]string {
  711. namespaceAnnotations := map[string]map[string]string{}
  712. for _, res := range resNamespaceAnnotations {
  713. namespace, err := res.GetString("namespace")
  714. if err != nil {
  715. continue
  716. }
  717. if _, ok := namespaceAnnotations[namespace]; !ok {
  718. namespaceAnnotations[namespace] = map[string]string{}
  719. }
  720. for k, l := range res.GetAnnotations() {
  721. namespaceAnnotations[namespace][k] = l
  722. }
  723. }
  724. return namespaceAnnotations
  725. }
  726. func resToPodAnnotations(resPodAnnotations []*prom.QueryResult) map[podKey]map[string]string {
  727. podAnnotations := map[podKey]map[string]string{}
  728. for _, res := range resPodAnnotations {
  729. podKey, err := resultPodKey(res, "cluster_id", "namespace", "pod")
  730. if err != nil {
  731. continue
  732. }
  733. if _, ok := podAnnotations[podKey]; !ok {
  734. podAnnotations[podKey] = map[string]string{}
  735. }
  736. for k, l := range res.GetAnnotations() {
  737. podAnnotations[podKey][k] = l
  738. }
  739. }
  740. return podAnnotations
  741. }
  742. func applyLabels(podMap map[podKey]*Pod, namespaceLabels map[string]map[string]string, podLabels map[podKey]map[string]string) {
  743. for key, pod := range podMap {
  744. for _, alloc := range pod.Allocations {
  745. allocLabels, err := alloc.Properties.GetLabels()
  746. if err != nil {
  747. allocLabels = map[string]string{}
  748. }
  749. // Apply namespace labels first, then pod labels so that pod labels
  750. // overwrite namespace labels.
  751. if labels, ok := namespaceLabels[key.Namespace]; ok {
  752. for k, v := range labels {
  753. allocLabels[k] = v
  754. }
  755. }
  756. if labels, ok := podLabels[key]; ok {
  757. for k, v := range labels {
  758. allocLabels[k] = v
  759. }
  760. }
  761. alloc.Properties.SetLabels(allocLabels)
  762. }
  763. }
  764. }
  765. func applyAnnotations(podMap map[podKey]*Pod, namespaceAnnotations map[string]map[string]string, podAnnotations map[podKey]map[string]string) {
  766. for key, pod := range podMap {
  767. for _, alloc := range pod.Allocations {
  768. allocAnnotations, err := alloc.Properties.GetAnnotations()
  769. if err != nil {
  770. allocAnnotations = map[string]string{}
  771. }
  772. // Apply namespace annotations first, then pod annotations so that
  773. // pod labels overwrite namespace labels.
  774. if labels, ok := namespaceAnnotations[key.Namespace]; ok {
  775. for k, v := range labels {
  776. allocAnnotations[k] = v
  777. }
  778. }
  779. if labels, ok := podAnnotations[key]; ok {
  780. for k, v := range labels {
  781. allocAnnotations[k] = v
  782. }
  783. }
  784. alloc.Properties.SetAnnotations(allocAnnotations)
  785. }
  786. }
  787. }
  788. func getServiceLabels(resServiceLabels []*prom.QueryResult) map[serviceKey]map[string]string {
  789. serviceLabels := map[serviceKey]map[string]string{}
  790. for _, res := range resServiceLabels {
  791. serviceKey, err := resultServiceKey(res, "cluster_id", "namespace", "service")
  792. if err != nil {
  793. continue
  794. }
  795. if _, ok := serviceLabels[serviceKey]; !ok {
  796. serviceLabels[serviceKey] = map[string]string{}
  797. }
  798. for k, l := range res.GetLabels() {
  799. serviceLabels[serviceKey][k] = l
  800. }
  801. }
  802. // Prune duplicate services. That is, if the same service exists with
  803. // hyphens instead of underscores, keep the one that uses hyphens.
  804. for key := range serviceLabels {
  805. if strings.Contains(key.Service, "_") {
  806. duplicateService := strings.Replace(key.Service, "_", "-", -1)
  807. duplicateKey := newServiceKey(key.Cluster, key.Namespace, duplicateService)
  808. if _, ok := serviceLabels[duplicateKey]; ok {
  809. delete(serviceLabels, key)
  810. }
  811. }
  812. }
  813. return serviceLabels
  814. }
  815. func resToDeploymentLabels(resDeploymentLabels []*prom.QueryResult) map[controllerKey]map[string]string {
  816. deploymentLabels := map[controllerKey]map[string]string{}
  817. for _, res := range resDeploymentLabels {
  818. controllerKey, err := resultDeploymentKey(res, "cluster_id", "namespace", "deployment")
  819. if err != nil {
  820. continue
  821. }
  822. if _, ok := deploymentLabels[controllerKey]; !ok {
  823. deploymentLabels[controllerKey] = map[string]string{}
  824. }
  825. for k, l := range res.GetLabels() {
  826. deploymentLabels[controllerKey][k] = l
  827. }
  828. }
  829. // Prune duplicate deployments. That is, if the same deployment exists with
  830. // hyphens instead of underscores, keep the one that uses hyphens.
  831. for key := range deploymentLabels {
  832. if strings.Contains(key.Controller, "_") {
  833. duplicateController := strings.Replace(key.Controller, "_", "-", -1)
  834. duplicateKey := newControllerKey(key.Cluster, key.Namespace, key.ControllerKind, duplicateController)
  835. if _, ok := deploymentLabels[duplicateKey]; ok {
  836. delete(deploymentLabels, key)
  837. }
  838. }
  839. }
  840. return deploymentLabels
  841. }
  842. func resToStatefulSetLabels(resStatefulSetLabels []*prom.QueryResult) map[controllerKey]map[string]string {
  843. statefulSetLabels := map[controllerKey]map[string]string{}
  844. for _, res := range resStatefulSetLabels {
  845. controllerKey, err := resultStatefulSetKey(res, "cluster_id", "namespace", "statefulSet")
  846. if err != nil {
  847. continue
  848. }
  849. if _, ok := statefulSetLabels[controllerKey]; !ok {
  850. statefulSetLabels[controllerKey] = map[string]string{}
  851. }
  852. for k, l := range res.GetLabels() {
  853. statefulSetLabels[controllerKey][k] = l
  854. }
  855. }
  856. // Prune duplicate stateful sets. That is, if the same stateful set exists
  857. // with hyphens instead of underscores, keep the one that uses hyphens.
  858. for key := range statefulSetLabels {
  859. if strings.Contains(key.Controller, "_") {
  860. duplicateController := strings.Replace(key.Controller, "_", "-", -1)
  861. duplicateKey := newControllerKey(key.Cluster, key.Namespace, key.ControllerKind, duplicateController)
  862. if _, ok := statefulSetLabels[duplicateKey]; ok {
  863. delete(statefulSetLabels, key)
  864. }
  865. }
  866. }
  867. return statefulSetLabels
  868. }
  869. func labelsToPodControllerMap(podLabels map[podKey]map[string]string, controllerLabels map[controllerKey]map[string]string) map[podKey]controllerKey {
  870. podControllerMap := map[podKey]controllerKey{}
  871. // For each controller, turn the labels into a selector and attempt to
  872. // match it with each set of pod labels. A match indicates that the pod
  873. // belongs to the controller.
  874. for cKey, cLabels := range controllerLabels {
  875. selector := labels.Set(cLabels).AsSelectorPreValidated()
  876. for pKey, pLabels := range podLabels {
  877. // If the pod is in a different cluster or namespace, there is
  878. // no need to compare the labels.
  879. if cKey.Cluster != pKey.Cluster || cKey.Namespace != pKey.Namespace {
  880. continue
  881. }
  882. podLabelSet := labels.Set(pLabels)
  883. if selector.Matches(podLabelSet) {
  884. if _, ok := podControllerMap[pKey]; ok {
  885. log.Warningf("CostModel.ComputeAllocation: PodControllerMap match already exists: %s matches %s and %s", pKey, podControllerMap[pKey], cKey)
  886. }
  887. podControllerMap[pKey] = cKey
  888. }
  889. }
  890. }
  891. return podControllerMap
  892. }
  893. func resToPodDaemonSetMap(resDaemonSetLabels []*prom.QueryResult) map[podKey]controllerKey {
  894. daemonSetLabels := map[podKey]controllerKey{}
  895. for _, res := range resDaemonSetLabels {
  896. controllerKey, err := resultDaemonSetKey(res, "cluster_id", "namespace", "owner_name")
  897. if err != nil {
  898. continue
  899. }
  900. pod, err := res.GetString("pod")
  901. if err != nil {
  902. log.Warningf("CostModel.ComputeAllocation: DaemonSetLabel result without pod: %s", controllerKey)
  903. }
  904. podKey := newPodKey(controllerKey.Cluster, controllerKey.Namespace, pod)
  905. daemonSetLabels[podKey] = controllerKey
  906. }
  907. return daemonSetLabels
  908. }
  909. func resToPodJobMap(resJobLabels []*prom.QueryResult) map[podKey]controllerKey {
  910. jobLabels := map[podKey]controllerKey{}
  911. for _, res := range resJobLabels {
  912. controllerKey, err := resultJobKey(res, "cluster_id", "namespace", "owner_name")
  913. if err != nil {
  914. continue
  915. }
  916. // Convert the name of Jobs generated by CronJobs to the name of the
  917. // CronJob by stripping the timestamp off the end.
  918. match := isCron.FindStringSubmatch(controllerKey.Controller)
  919. if match != nil {
  920. controllerKey.Controller = match[1]
  921. }
  922. pod, err := res.GetString("pod")
  923. if err != nil {
  924. log.Warningf("CostModel.ComputeAllocation: JobLabel result without pod: %s", controllerKey)
  925. }
  926. podKey := newPodKey(controllerKey.Cluster, controllerKey.Namespace, pod)
  927. jobLabels[podKey] = controllerKey
  928. }
  929. return jobLabels
  930. }
  931. func applyServicesToPods(podMap map[podKey]*Pod, podLabels map[podKey]map[string]string, allocsByService map[serviceKey][]*kubecost.Allocation, serviceLabels map[serviceKey]map[string]string) {
  932. podServicesMap := map[podKey][]serviceKey{}
  933. // For each service, turn the labels into a selector and attempt to
  934. // match it with each set of pod labels. A match indicates that the pod
  935. // belongs to the service.
  936. for sKey, sLabels := range serviceLabels {
  937. selector := labels.Set(sLabels).AsSelectorPreValidated()
  938. for pKey, pLabels := range podLabels {
  939. // If the pod is in a different cluster or namespace, there is
  940. // no need to compare the labels.
  941. if sKey.Cluster != pKey.Cluster || sKey.Namespace != pKey.Namespace {
  942. continue
  943. }
  944. podLabelSet := labels.Set(pLabels)
  945. if selector.Matches(podLabelSet) {
  946. if _, ok := podServicesMap[pKey]; !ok {
  947. podServicesMap[pKey] = []serviceKey{}
  948. }
  949. podServicesMap[pKey] = append(podServicesMap[pKey], sKey)
  950. }
  951. }
  952. }
  953. // For each allocation in each pod, attempt to find and apply the list of
  954. // services associated with the allocation's pod.
  955. for key, pod := range podMap {
  956. for _, alloc := range pod.Allocations {
  957. if sKeys, ok := podServicesMap[key]; ok {
  958. services := []string{}
  959. for _, sKey := range sKeys {
  960. services = append(services, sKey.Service)
  961. allocsByService[sKey] = append(allocsByService[sKey], alloc)
  962. }
  963. alloc.Properties.SetServices(services)
  964. }
  965. }
  966. }
  967. }
  968. func applyControllersToPods(podMap map[podKey]*Pod, podControllerMap map[podKey]controllerKey) {
  969. for key, pod := range podMap {
  970. for _, alloc := range pod.Allocations {
  971. if controllerKey, ok := podControllerMap[key]; ok {
  972. alloc.Properties.SetControllerKind(controllerKey.ControllerKind)
  973. alloc.Properties.SetController(controllerKey.Controller)
  974. }
  975. }
  976. }
  977. }
  978. func applyNodeCostPerCPUHr(nodeMap map[nodeKey]*NodePricing, resNodeCostPerCPUHr []*prom.QueryResult) {
  979. for _, res := range resNodeCostPerCPUHr {
  980. cluster, err := res.GetString("cluster_id")
  981. if err != nil {
  982. cluster = env.GetClusterID()
  983. }
  984. node, err := res.GetString("node")
  985. if err != nil {
  986. log.Warningf("CostModel.ComputeAllocation: Node CPU cost query result missing field: %s", err)
  987. continue
  988. }
  989. instanceType, err := res.GetString("instance_type")
  990. if err != nil {
  991. log.Warningf("CostModel.ComputeAllocation: Node CPU cost query result missing field: %s", err)
  992. continue
  993. }
  994. key := newNodeKey(cluster, node)
  995. if _, ok := nodeMap[key]; !ok {
  996. nodeMap[key] = &NodePricing{
  997. Name: node,
  998. NodeType: instanceType,
  999. }
  1000. }
  1001. nodeMap[key].CostPerCPUHr = res.Values[0].Value
  1002. }
  1003. }
  1004. func applyNodeCostPerRAMGiBHr(nodeMap map[nodeKey]*NodePricing, resNodeCostPerRAMGiBHr []*prom.QueryResult) {
  1005. for _, res := range resNodeCostPerRAMGiBHr {
  1006. cluster, err := res.GetString("cluster_id")
  1007. if err != nil {
  1008. cluster = env.GetClusterID()
  1009. }
  1010. node, err := res.GetString("node")
  1011. if err != nil {
  1012. log.Warningf("CostModel.ComputeAllocation: Node RAM cost query result missing field: %s", err)
  1013. continue
  1014. }
  1015. instanceType, err := res.GetString("instance_type")
  1016. if err != nil {
  1017. log.Warningf("CostModel.ComputeAllocation: Node RAM cost query result missing field: %s", err)
  1018. continue
  1019. }
  1020. key := newNodeKey(cluster, node)
  1021. if _, ok := nodeMap[key]; !ok {
  1022. nodeMap[key] = &NodePricing{
  1023. Name: node,
  1024. NodeType: instanceType,
  1025. }
  1026. }
  1027. nodeMap[key].CostPerRAMGiBHr = res.Values[0].Value
  1028. }
  1029. }
  1030. func applyNodeCostPerGPUHr(nodeMap map[nodeKey]*NodePricing, resNodeCostPerGPUHr []*prom.QueryResult) {
  1031. for _, res := range resNodeCostPerGPUHr {
  1032. cluster, err := res.GetString("cluster_id")
  1033. if err != nil {
  1034. cluster = env.GetClusterID()
  1035. }
  1036. node, err := res.GetString("node")
  1037. if err != nil {
  1038. log.Warningf("CostModel.ComputeAllocation: Node GPU cost query result missing field: %s", err)
  1039. continue
  1040. }
  1041. instanceType, err := res.GetString("instance_type")
  1042. if err != nil {
  1043. log.Warningf("CostModel.ComputeAllocation: Node GPU cost query result missing field: %s", err)
  1044. continue
  1045. }
  1046. key := newNodeKey(cluster, node)
  1047. if _, ok := nodeMap[key]; !ok {
  1048. nodeMap[key] = &NodePricing{
  1049. Name: node,
  1050. NodeType: instanceType,
  1051. }
  1052. }
  1053. nodeMap[key].CostPerGPUHr = res.Values[0].Value
  1054. }
  1055. }
  1056. func applyNodeSpot(nodeMap map[nodeKey]*NodePricing, resNodeIsSpot []*prom.QueryResult) {
  1057. for _, res := range resNodeIsSpot {
  1058. cluster, err := res.GetString("cluster_id")
  1059. if err != nil {
  1060. cluster = env.GetClusterID()
  1061. }
  1062. node, err := res.GetString("node")
  1063. if err != nil {
  1064. log.Warningf("CostModel.ComputeAllocation: Node spot query result missing field: %s", err)
  1065. continue
  1066. }
  1067. key := newNodeKey(cluster, node)
  1068. if _, ok := nodeMap[key]; !ok {
  1069. log.Warningf("CostModel.ComputeAllocation: Node spot query result for missing node: %s", key)
  1070. continue
  1071. }
  1072. nodeMap[key].Preemptible = res.Values[0].Value > 0
  1073. }
  1074. }
  1075. func applyNodeDiscount(nodeMap map[nodeKey]*NodePricing, cm *CostModel) {
  1076. if cm == nil {
  1077. return
  1078. }
  1079. c, err := cm.Provider.GetConfig()
  1080. if err != nil {
  1081. log.Errorf("CostModel.ComputeAllocation: applyNodeDiscount: %s", err)
  1082. return
  1083. }
  1084. discount, err := ParsePercentString(c.Discount)
  1085. if err != nil {
  1086. log.Errorf("CostModel.ComputeAllocation: applyNodeDiscount: %s", err)
  1087. return
  1088. }
  1089. negotiatedDiscount, err := ParsePercentString(c.NegotiatedDiscount)
  1090. if err != nil {
  1091. log.Errorf("CostModel.ComputeAllocation: applyNodeDiscount: %s", err)
  1092. return
  1093. }
  1094. for _, node := range nodeMap {
  1095. // TODO GKE Reserved Instances into account
  1096. node.Discount = cm.Provider.CombinedDiscountForNode(node.NodeType, node.Preemptible, discount, negotiatedDiscount)
  1097. node.CostPerCPUHr *= (1.0 - node.Discount)
  1098. node.CostPerRAMGiBHr *= (1.0 - node.Discount)
  1099. }
  1100. }
  1101. func buildPVMap(pvMap map[pvKey]*PV, resPVCostPerGiBHour []*prom.QueryResult) {
  1102. for _, res := range resPVCostPerGiBHour {
  1103. cluster, err := res.GetString("cluster_id")
  1104. if err != nil {
  1105. cluster = env.GetClusterID()
  1106. }
  1107. name, err := res.GetString("volumename")
  1108. if err != nil {
  1109. log.Warningf("CostModel.ComputeAllocation: PV cost without volumename")
  1110. continue
  1111. }
  1112. key := newPVKey(cluster, name)
  1113. pvMap[key] = &PV{
  1114. Cluster: cluster,
  1115. Name: name,
  1116. CostPerGiBHour: res.Values[0].Value,
  1117. }
  1118. }
  1119. }
  1120. func applyPVBytes(pvMap map[pvKey]*PV, resPVBytes []*prom.QueryResult) {
  1121. for _, res := range resPVBytes {
  1122. key, err := resultPVKey(res, "cluster_id", "persistentvolume")
  1123. if err != nil {
  1124. log.Warningf("CostModel.ComputeAllocation: PV bytes query result missing field: %s", err)
  1125. continue
  1126. }
  1127. if _, ok := pvMap[key]; !ok {
  1128. log.Warningf("CostModel.ComputeAllocation: PV bytes result for missing PV: %s", err)
  1129. continue
  1130. }
  1131. pvMap[key].Bytes = res.Values[0].Value
  1132. }
  1133. }
  1134. func buildPVCMap(window kubecost.Window, pvcMap map[pvcKey]*PVC, pvMap map[pvKey]*PV, resPVCInfo []*prom.QueryResult) {
  1135. for _, res := range resPVCInfo {
  1136. cluster, err := res.GetString("cluster_id")
  1137. if err != nil {
  1138. cluster = env.GetClusterID()
  1139. }
  1140. values, err := res.GetStrings("persistentvolumeclaim", "storageclass", "volumename", "namespace")
  1141. if err != nil {
  1142. log.DedupedWarningf(10, "CostModel.ComputeAllocation: PVC info query result missing field: %s", err)
  1143. continue
  1144. }
  1145. namespace := values["namespace"]
  1146. name := values["persistentvolumeclaim"]
  1147. volume := values["volumename"]
  1148. storageClass := values["storageclass"]
  1149. pvKey := newPVKey(cluster, volume)
  1150. pvcKey := newPVCKey(cluster, namespace, name)
  1151. // pvcStart and pvcEnd are the timestamps of the first and last minutes
  1152. // the PVC was running, respectively. We subtract 1m from pvcStart
  1153. // because this point will actually represent the end of the first
  1154. // minute. We don't subtract from pvcEnd because it already represents
  1155. // the end of the last minute.
  1156. var pvcStart, pvcEnd time.Time
  1157. for _, datum := range res.Values {
  1158. t := time.Unix(int64(datum.Timestamp), 0)
  1159. if pvcStart.IsZero() && datum.Value > 0 && window.Contains(t) {
  1160. pvcStart = t
  1161. }
  1162. if datum.Value > 0 && window.Contains(t) {
  1163. pvcEnd = t
  1164. }
  1165. }
  1166. if pvcStart.IsZero() || pvcEnd.IsZero() {
  1167. log.Warningf("CostModel.ComputeAllocation: PVC %s has no running time", pvcKey)
  1168. }
  1169. pvcStart = pvcStart.Add(-time.Minute)
  1170. if _, ok := pvMap[pvKey]; !ok {
  1171. continue
  1172. }
  1173. pvMap[pvKey].StorageClass = storageClass
  1174. if _, ok := pvcMap[pvcKey]; !ok {
  1175. pvcMap[pvcKey] = &PVC{}
  1176. }
  1177. pvcMap[pvcKey].Name = name
  1178. pvcMap[pvcKey].Namespace = namespace
  1179. pvcMap[pvcKey].Volume = pvMap[pvKey]
  1180. pvcMap[pvcKey].Start = pvcStart
  1181. pvcMap[pvcKey].End = pvcEnd
  1182. }
  1183. }
  1184. func applyPVCBytesRequested(pvcMap map[pvcKey]*PVC, resPVCBytesRequested []*prom.QueryResult) {
  1185. for _, res := range resPVCBytesRequested {
  1186. key, err := resultPVCKey(res, "cluster_id", "namespace", "persistentvolumeclaim")
  1187. if err != nil {
  1188. continue
  1189. }
  1190. if _, ok := pvcMap[key]; !ok {
  1191. continue
  1192. }
  1193. pvcMap[key].Bytes = res.Values[0].Value
  1194. }
  1195. }
  1196. func buildPodPVCMap(podPVCMap map[podKey][]*PVC, pvMap map[pvKey]*PV, pvcMap map[pvcKey]*PVC, podMap map[podKey]*Pod, resPodPVCAllocation []*prom.QueryResult) {
  1197. for _, res := range resPodPVCAllocation {
  1198. cluster, err := res.GetString("cluster_id")
  1199. if err != nil {
  1200. cluster = env.GetClusterID()
  1201. }
  1202. values, err := res.GetStrings("persistentvolume", "persistentvolumeclaim", "pod", "namespace")
  1203. if err != nil {
  1204. log.Warningf("CostModel.ComputeAllocation: PVC allocation query result missing field: %s", err)
  1205. continue
  1206. }
  1207. namespace := values["namespace"]
  1208. pod := values["pod"]
  1209. name := values["persistentvolumeclaim"]
  1210. volume := values["persistentvolume"]
  1211. podKey := newPodKey(cluster, namespace, pod)
  1212. pvKey := newPVKey(cluster, volume)
  1213. pvcKey := newPVCKey(cluster, namespace, name)
  1214. if _, ok := pvMap[pvKey]; !ok {
  1215. log.Warningf("CostModel.ComputeAllocation: PV missing for PVC allocation query result: %s", pvKey)
  1216. continue
  1217. }
  1218. if _, ok := podPVCMap[podKey]; !ok {
  1219. podPVCMap[podKey] = []*PVC{}
  1220. }
  1221. pvc, ok := pvcMap[pvcKey]
  1222. if !ok {
  1223. log.Warningf("CostModel.ComputeAllocation: PVC missing for PVC allocation query: %s", pvcKey)
  1224. continue
  1225. }
  1226. count := 1
  1227. if pod, ok := podMap[podKey]; ok && len(pod.Allocations) > 0 {
  1228. count = len(pod.Allocations)
  1229. } else {
  1230. log.DedupedWarningf(10, "CostModel.ComputeAllocation: PVC %s for missing pod %s", pvcKey, podKey)
  1231. }
  1232. pvc.Count = count
  1233. pvc.Mounted = true
  1234. podPVCMap[podKey] = append(podPVCMap[podKey], pvc)
  1235. }
  1236. }
  1237. func applyUnmountedPVs(window kubecost.Window, podMap map[podKey]*Pod, pvMap map[pvKey]*PV, pvcMap map[pvcKey]*PVC) {
  1238. unmountedPVBytes := map[string]float64{}
  1239. unmountedPVCost := map[string]float64{}
  1240. for _, pv := range pvMap {
  1241. mounted := false
  1242. for _, pvc := range pvcMap {
  1243. if pvc.Volume == nil {
  1244. continue
  1245. }
  1246. if pvc.Volume == pv {
  1247. mounted = true
  1248. break
  1249. }
  1250. }
  1251. if !mounted {
  1252. gib := pv.Bytes / 1024 / 1024 / 1024
  1253. hrs := window.Minutes() / 60.0 // TODO improve with PV hours, not window hours
  1254. cost := pv.CostPerGiBHour * gib * hrs
  1255. unmountedPVCost[pv.Cluster] += cost
  1256. unmountedPVBytes[pv.Cluster] += pv.Bytes
  1257. }
  1258. }
  1259. for cluster, amount := range unmountedPVCost {
  1260. container := kubecost.UnmountedSuffix
  1261. pod := kubecost.UnmountedSuffix
  1262. namespace := kubecost.UnmountedSuffix
  1263. node := ""
  1264. key := newPodKey(cluster, namespace, pod)
  1265. podMap[key] = &Pod{
  1266. Window: window.Clone(),
  1267. Start: *window.Start(),
  1268. End: *window.End(),
  1269. Key: key,
  1270. Allocations: map[string]*kubecost.Allocation{},
  1271. }
  1272. podMap[key].AppendContainer(container)
  1273. podMap[key].Allocations[container].Properties.SetCluster(cluster)
  1274. podMap[key].Allocations[container].Properties.SetNode(node)
  1275. podMap[key].Allocations[container].Properties.SetNamespace(namespace)
  1276. podMap[key].Allocations[container].Properties.SetPod(pod)
  1277. podMap[key].Allocations[container].Properties.SetContainer(container)
  1278. podMap[key].Allocations[container].PVByteHours = unmountedPVBytes[cluster] * window.Minutes() / 60.0
  1279. podMap[key].Allocations[container].PVCost = amount
  1280. }
  1281. }
  1282. func applyUnmountedPVCs(window kubecost.Window, podMap map[podKey]*Pod, pvcMap map[pvcKey]*PVC) {
  1283. unmountedPVCBytes := map[namespaceKey]float64{}
  1284. unmountedPVCCost := map[namespaceKey]float64{}
  1285. for _, pvc := range pvcMap {
  1286. if !pvc.Mounted && pvc.Volume != nil {
  1287. key := newNamespaceKey(pvc.Cluster, pvc.Namespace)
  1288. gib := pvc.Volume.Bytes / 1024 / 1024 / 1024
  1289. hrs := pvc.Minutes() / 60.0
  1290. cost := pvc.Volume.CostPerGiBHour * gib * hrs
  1291. unmountedPVCCost[key] += cost
  1292. unmountedPVCBytes[key] += pvc.Volume.Bytes
  1293. }
  1294. }
  1295. for key, amount := range unmountedPVCCost {
  1296. container := kubecost.UnmountedSuffix
  1297. pod := kubecost.UnmountedSuffix
  1298. namespace := key.Namespace
  1299. node := ""
  1300. cluster := key.Cluster
  1301. podKey := newPodKey(cluster, namespace, pod)
  1302. podMap[podKey] = &Pod{
  1303. Window: window.Clone(),
  1304. Start: *window.Start(),
  1305. End: *window.End(),
  1306. Key: podKey,
  1307. Allocations: map[string]*kubecost.Allocation{},
  1308. }
  1309. podMap[podKey].AppendContainer(container)
  1310. podMap[podKey].Allocations[container].Properties.SetCluster(cluster)
  1311. podMap[podKey].Allocations[container].Properties.SetNode(node)
  1312. podMap[podKey].Allocations[container].Properties.SetNamespace(namespace)
  1313. podMap[podKey].Allocations[container].Properties.SetPod(pod)
  1314. podMap[podKey].Allocations[container].Properties.SetContainer(container)
  1315. podMap[podKey].Allocations[container].PVByteHours = unmountedPVCBytes[key] * window.Minutes() / 60.0
  1316. podMap[podKey].Allocations[container].PVCost = amount
  1317. }
  1318. }
  1319. // LB describes the start and end time of a Load Balancer along with cost
  1320. type LB struct {
  1321. TotalCost float64
  1322. Start time.Time
  1323. End time.Time
  1324. }
  1325. func getLoadBalancerCosts(resLBCost, resLBActiveMins []*prom.QueryResult, resolution time.Duration) map[serviceKey]*LB {
  1326. lbMap := make(map[serviceKey]*LB)
  1327. lbHourlyCosts := make(map[serviceKey]float64)
  1328. for _, res := range resLBCost {
  1329. serviceKey, err := resultServiceKey(res, "cluster_id", "namespace", "service_name")
  1330. if err != nil {
  1331. continue
  1332. }
  1333. lbHourlyCosts[serviceKey] = res.Values[0].Value
  1334. }
  1335. for _, res := range resLBActiveMins {
  1336. serviceKey, err := resultServiceKey(res, "cluster_id", "namespace", "service_name")
  1337. if err != nil || len(res.Values) == 0 {
  1338. continue
  1339. }
  1340. if _, ok := lbHourlyCosts[serviceKey]; !ok {
  1341. log.Warningf("CostModel: failed to find hourly cost for Load Balancer: %v", serviceKey)
  1342. continue
  1343. }
  1344. s := time.Unix(int64(res.Values[0].Timestamp), 0)
  1345. // subtract resolution from start time to cover full time period
  1346. s = s.Add(-resolution)
  1347. e := time.Unix(int64(res.Values[len(res.Values)-1].Timestamp), 0)
  1348. hours := e.Sub(s).Hours()
  1349. lbMap[serviceKey] = &LB{
  1350. TotalCost: lbHourlyCosts[serviceKey] * hours,
  1351. Start: s,
  1352. End: e,
  1353. }
  1354. }
  1355. return lbMap
  1356. }
  1357. func applyLoadBalancersToPods(lbMap map[serviceKey]*LB, allocsByService map[serviceKey][]*kubecost.Allocation) {
  1358. for sKey, lb := range lbMap {
  1359. totalHours := 0.0
  1360. allocHours := make(map[*kubecost.Allocation]float64)
  1361. // Add portion of load balancing cost to each allocation
  1362. // proportional to the total number of hours allocations used the load balancer
  1363. for _, alloc := range allocsByService[sKey] {
  1364. // Determine the (start, end) of the relationship between the
  1365. // given LB and the associated Allocation so that a precise
  1366. // number of hours can be used to compute cumulative cost.
  1367. s, e := alloc.Start, alloc.End
  1368. if lb.Start.After(alloc.Start) {
  1369. s = lb.Start
  1370. }
  1371. if lb.End.Before(alloc.End) {
  1372. e = lb.End
  1373. }
  1374. hours := e.Sub(s).Hours()
  1375. // A negative number of hours signifies no overlap between the windows
  1376. if hours < 0 {
  1377. totalHours += hours
  1378. allocHours[alloc] = hours
  1379. }
  1380. }
  1381. // Distribute cost of service once total hours is calculated
  1382. for alloc, hours := range allocHours {
  1383. alloc.NetworkCost += lb.TotalCost * hours / totalHours
  1384. }
  1385. }
  1386. }
  1387. // getNodePricing determines node pricing, given a key and a mapping from keys
  1388. // to their NodePricing instances, as well as the custom pricing configuration
  1389. // inherent to the CostModel instance. If custom pricing is set, use that. If
  1390. // not, use the pricing defined by the given key. If that doesn't exist, fall
  1391. // back on custom pricing as a default.
  1392. func (cm *CostModel) getNodePricing(nodeMap map[nodeKey]*NodePricing, nodeKey nodeKey) *NodePricing {
  1393. // Find the relevant NodePricing, if it exists. If not, substitute the
  1394. // custom NodePricing as a default.
  1395. node, ok := nodeMap[nodeKey]
  1396. if !ok || node == nil {
  1397. if nodeKey.Node != "" {
  1398. log.Warningf("CostModel: failed to find node for %s", nodeKey)
  1399. }
  1400. return cm.getCustomNodePricing(false)
  1401. }
  1402. // If custom pricing is enabled and can be retrieved, override detected
  1403. // node pricing with the custom values.
  1404. customPricingConfig, err := cm.Provider.GetConfig()
  1405. if err != nil {
  1406. log.Warningf("CostModel: failed to load custom pricing: %s", err)
  1407. }
  1408. if cloud.CustomPricesEnabled(cm.Provider) && customPricingConfig != nil {
  1409. return cm.getCustomNodePricing(node.Preemptible)
  1410. }
  1411. node.Source = "prometheus"
  1412. // If any of the values are NaN or zero, replace them with the custom
  1413. // values as default.
  1414. // TODO:CLEANUP can't we parse these custom prices once? why do we store
  1415. // them as strings like this?
  1416. if node.CostPerCPUHr == 0 || math.IsNaN(node.CostPerCPUHr) {
  1417. log.Warningf("CostModel: node pricing has illegal CostPerCPUHr; replacing with custom pricing: %s", nodeKey)
  1418. cpuCostStr := customPricingConfig.CPU
  1419. if node.Preemptible {
  1420. cpuCostStr = customPricingConfig.SpotCPU
  1421. }
  1422. costPerCPUHr, err := strconv.ParseFloat(cpuCostStr, 64)
  1423. if err != nil {
  1424. log.Warningf("CostModel: custom pricing has illegal CPU cost: %s", cpuCostStr)
  1425. }
  1426. node.CostPerCPUHr = costPerCPUHr
  1427. node.Source += "/customCPU"
  1428. }
  1429. if math.IsNaN(node.CostPerGPUHr) {
  1430. log.Warningf("CostModel: node pricing has illegal CostPerGPUHr; replacing with custom pricing: %s", nodeKey)
  1431. gpuCostStr := customPricingConfig.GPU
  1432. if node.Preemptible {
  1433. gpuCostStr = customPricingConfig.SpotGPU
  1434. }
  1435. costPerGPUHr, err := strconv.ParseFloat(gpuCostStr, 64)
  1436. if err != nil {
  1437. log.Warningf("CostModel: custom pricing has illegal GPU cost: %s", gpuCostStr)
  1438. }
  1439. node.CostPerGPUHr = costPerGPUHr
  1440. node.Source += "/customGPU"
  1441. }
  1442. if node.CostPerRAMGiBHr == 0 || math.IsNaN(node.CostPerRAMGiBHr) {
  1443. log.Warningf("CostModel: node pricing has illegal CostPerRAMHr; replacing with custom pricing: %s", nodeKey)
  1444. ramCostStr := customPricingConfig.RAM
  1445. if node.Preemptible {
  1446. ramCostStr = customPricingConfig.SpotRAM
  1447. }
  1448. costPerRAMHr, err := strconv.ParseFloat(ramCostStr, 64)
  1449. if err != nil {
  1450. log.Warningf("CostModel: custom pricing has illegal RAM cost: %s", ramCostStr)
  1451. }
  1452. node.CostPerRAMGiBHr = costPerRAMHr
  1453. node.Source += "/customRAM"
  1454. }
  1455. return node
  1456. }
  1457. // getCustomNodePricing converts the CostModel's configured custom pricing
  1458. // values into a NodePricing instance.
  1459. func (cm *CostModel) getCustomNodePricing(spot bool) *NodePricing {
  1460. customPricingConfig, err := cm.Provider.GetConfig()
  1461. if err != nil {
  1462. return nil
  1463. }
  1464. cpuCostStr := customPricingConfig.CPU
  1465. gpuCostStr := customPricingConfig.GPU
  1466. ramCostStr := customPricingConfig.RAM
  1467. if spot {
  1468. cpuCostStr = customPricingConfig.SpotCPU
  1469. gpuCostStr = customPricingConfig.SpotGPU
  1470. ramCostStr = customPricingConfig.SpotRAM
  1471. }
  1472. node := &NodePricing{Source: "custom"}
  1473. costPerCPUHr, err := strconv.ParseFloat(cpuCostStr, 64)
  1474. if err != nil {
  1475. log.Warningf("CostModel: custom pricing has illegal CPU cost: %s", cpuCostStr)
  1476. }
  1477. node.CostPerCPUHr = costPerCPUHr
  1478. costPerGPUHr, err := strconv.ParseFloat(gpuCostStr, 64)
  1479. if err != nil {
  1480. log.Warningf("CostModel: custom pricing has illegal GPU cost: %s", gpuCostStr)
  1481. }
  1482. node.CostPerGPUHr = costPerGPUHr
  1483. costPerRAMHr, err := strconv.ParseFloat(ramCostStr, 64)
  1484. if err != nil {
  1485. log.Warningf("CostModel: custom pricing has illegal RAM cost: %s", ramCostStr)
  1486. }
  1487. node.CostPerRAMGiBHr = costPerRAMHr
  1488. return node
  1489. }
  1490. // NodePricing describes the resource costs associated with a given node, as
  1491. // well as the source of the information (e.g. prometheus, custom)
  1492. type NodePricing struct {
  1493. Name string
  1494. NodeType string
  1495. Preemptible bool
  1496. CostPerCPUHr float64
  1497. CostPerRAMGiBHr float64
  1498. CostPerGPUHr float64
  1499. Discount float64
  1500. Source string
  1501. }
  1502. // Pod describes a running pod's start and end time within a Window and
  1503. // all the Allocations (i.e. containers) contained within it.
  1504. type Pod struct {
  1505. Window kubecost.Window
  1506. Start time.Time
  1507. End time.Time
  1508. Key podKey
  1509. Allocations map[string]*kubecost.Allocation
  1510. }
  1511. // AppendContainer adds an entry for the given container name to the Pod.
  1512. func (p Pod) AppendContainer(container string) {
  1513. name := fmt.Sprintf("%s/%s/%s/%s", p.Key.Cluster, p.Key.Namespace, p.Key.Pod, container)
  1514. alloc := &kubecost.Allocation{
  1515. Name: name,
  1516. Properties: kubecost.Properties{},
  1517. Window: p.Window.Clone(),
  1518. Start: p.Start,
  1519. End: p.End,
  1520. }
  1521. alloc.Properties.SetContainer(container)
  1522. alloc.Properties.SetPod(p.Key.Pod)
  1523. alloc.Properties.SetNamespace(p.Key.Namespace)
  1524. alloc.Properties.SetCluster(p.Key.Cluster)
  1525. p.Allocations[container] = alloc
  1526. }
  1527. // PVC describes a PersistentVolumeClaim
  1528. // TODO:CLEANUP move to pkg/kubecost?
  1529. // TODO:CLEANUP add PersistentVolumeClaims field to type Allocation?
  1530. type PVC struct {
  1531. Bytes float64 `json:"bytes"`
  1532. Count int `json:"count"`
  1533. Name string `json:"name"`
  1534. Cluster string `json:"cluster"`
  1535. Namespace string `json:"namespace"`
  1536. Volume *PV `json:"persistentVolume"`
  1537. Mounted bool `json:"mounted"`
  1538. Start time.Time `json:"start"`
  1539. End time.Time `json:"end"`
  1540. }
  1541. // Cost computes the cumulative cost of the PVC
  1542. func (pvc *PVC) Cost() float64 {
  1543. if pvc == nil || pvc.Volume == nil {
  1544. return 0.0
  1545. }
  1546. gib := pvc.Bytes / 1024 / 1024 / 1024
  1547. hrs := pvc.Minutes() / 60.0
  1548. return pvc.Volume.CostPerGiBHour * gib * hrs
  1549. }
  1550. // Minutes computes the number of minutes over which the PVC is defined
  1551. func (pvc *PVC) Minutes() float64 {
  1552. if pvc == nil {
  1553. return 0.0
  1554. }
  1555. return pvc.End.Sub(pvc.Start).Minutes()
  1556. }
  1557. // String returns a string representation of the PVC
  1558. func (pvc *PVC) String() string {
  1559. if pvc == nil {
  1560. return "<nil>"
  1561. }
  1562. 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))
  1563. }
  1564. // PV describes a PersistentVolume
  1565. // TODO:CLEANUP move to pkg/kubecost?
  1566. type PV struct {
  1567. Bytes float64 `json:"bytes"`
  1568. CostPerGiBHour float64 `json:"costPerGiBHour"`
  1569. Cluster string `json:"cluster"`
  1570. Name string `json:"name"`
  1571. StorageClass string `json:"storageClass"`
  1572. }
  1573. // String returns a string representation of the PV
  1574. func (pv *PV) String() string {
  1575. if pv == nil {
  1576. return "<nil>"
  1577. }
  1578. return fmt.Sprintf("%s/%s{Bytes:%.2f, Cost/GiB*Hr:%.6f, StorageClass:%s}", pv.Cluster, pv.Name, pv.Bytes, pv.CostPerGiBHour, pv.StorageClass)
  1579. }