allocation_helpers.go 68 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236
  1. package costmodel
  2. import (
  3. "fmt"
  4. "math"
  5. "strconv"
  6. "strings"
  7. "time"
  8. "github.com/opencost/opencost/pkg/cloud/provider"
  9. "github.com/opencost/opencost/pkg/env"
  10. "github.com/opencost/opencost/pkg/kubecost"
  11. "github.com/opencost/opencost/pkg/log"
  12. "github.com/opencost/opencost/pkg/prom"
  13. "github.com/opencost/opencost/pkg/util/timeutil"
  14. "k8s.io/apimachinery/pkg/labels"
  15. )
  16. // This is a bit of a hack to work around garbage data from cadvisor
  17. // Ideally you cap each pod to the max CPU on its node, but that involves a bit more complexity, as it it would need to be done when allocations joins with asset data.
  18. const MAX_CPU_CAP = 512
  19. /* Pod Helpers */
  20. func (cm *CostModel) buildPodMap(window kubecost.Window, resolution, maxBatchSize time.Duration, podMap map[podKey]*pod, clusterStart, clusterEnd map[string]time.Time, ingestPodUID bool, podUIDKeyMap map[podKey][]podKey) error {
  21. // Assumes that window is positive and closed
  22. start, end := *window.Start(), *window.End()
  23. // Convert resolution duration to a query-ready string
  24. resStr := timeutil.DurationString(resolution)
  25. ctx := prom.NewNamedContext(cm.PrometheusClient, prom.AllocationContextName)
  26. // Query for (start, end) by (pod, namespace, cluster) over the given
  27. // window, using the given resolution, and if necessary in batches no
  28. // larger than the given maximum batch size. If working in batches, track
  29. // overall progress by starting with (window.start, window.start) and
  30. // querying in batches no larger than maxBatchSize from start-to-end,
  31. // folding each result set into podMap as the results come back.
  32. coverage := kubecost.NewWindow(&start, &start)
  33. numQuery := 1
  34. for coverage.End().Before(end) {
  35. // Determine the (start, end) of the current batch
  36. batchStart := *coverage.End()
  37. batchEnd := coverage.End().Add(maxBatchSize)
  38. if batchEnd.After(end) {
  39. batchEnd = end
  40. }
  41. var resPods []*prom.QueryResult
  42. var err error
  43. maxTries := 3
  44. numTries := 0
  45. for resPods == nil && numTries < maxTries {
  46. numTries++
  47. // Query for the duration between start and end
  48. durStr := timeutil.DurationString(batchEnd.Sub(batchStart))
  49. if durStr == "" {
  50. // Negative duration, so set empty results and don't query
  51. resPods = []*prom.QueryResult{}
  52. err = nil
  53. break
  54. }
  55. // Submit and profile query
  56. var queryPods string
  57. // If ingesting UIDs, avg on them
  58. if ingestPodUID {
  59. queryPods = fmt.Sprintf(queryFmtPodsUID, env.GetPromClusterFilter(), env.GetPromClusterLabel(), durStr, resStr)
  60. } else {
  61. queryPods = fmt.Sprintf(queryFmtPods, env.GetPromClusterFilter(), env.GetPromClusterLabel(), durStr, resStr)
  62. }
  63. queryProfile := time.Now()
  64. resPods, err = ctx.QueryAtTime(queryPods, batchEnd).Await()
  65. if err != nil {
  66. log.Profile(queryProfile, fmt.Sprintf("CostModel.ComputeAllocation: pod query %d try %d failed: %s", numQuery, numTries, queryPods))
  67. resPods = nil
  68. }
  69. }
  70. if err != nil {
  71. return err
  72. }
  73. // queryFmtPodsUID will return both UID-containing results, and non-UID-containing results,
  74. // so filter out the non-containing results so we don't duplicate pods. This is due to the
  75. // default setup of Kubecost having replicated kube_pod_container_status_running and
  76. // included KSM kube_pod_container_status_running. Querying w/ UID will return both.
  77. if ingestPodUID {
  78. var resPodsUID []*prom.QueryResult
  79. for _, res := range resPods {
  80. _, err := res.GetString("uid")
  81. if err == nil {
  82. resPodsUID = append(resPodsUID, res)
  83. }
  84. }
  85. if len(resPodsUID) > 0 {
  86. resPods = resPodsUID
  87. } else {
  88. log.DedupedWarningf(5, "CostModel.ComputeAllocation: UID ingestion enabled, but query did not return any results with UID")
  89. }
  90. }
  91. applyPodResults(window, resolution, podMap, clusterStart, clusterEnd, resPods, ingestPodUID, podUIDKeyMap)
  92. coverage = coverage.ExpandEnd(batchEnd)
  93. numQuery++
  94. }
  95. return nil
  96. }
  97. func applyPodResults(window kubecost.Window, resolution time.Duration, podMap map[podKey]*pod, clusterStart, clusterEnd map[string]time.Time, resPods []*prom.QueryResult, ingestPodUID bool, podUIDKeyMap map[podKey][]podKey) {
  98. for _, res := range resPods {
  99. if len(res.Values) == 0 {
  100. log.Warnf("CostModel.ComputeAllocation: empty minutes result")
  101. continue
  102. }
  103. cluster, err := res.GetString(env.GetPromClusterLabel())
  104. if err != nil {
  105. cluster = env.GetClusterID()
  106. }
  107. labels, err := res.GetStrings("namespace", "pod")
  108. if err != nil {
  109. log.Warnf("CostModel.ComputeAllocation: minutes query result missing field: %s", err)
  110. continue
  111. }
  112. namespace := labels["namespace"]
  113. podName := labels["pod"]
  114. key := newPodKey(cluster, namespace, podName)
  115. // If thisPod UIDs are being used to ID pods, append them to the thisPod name in
  116. // the podKey.
  117. if ingestPodUID {
  118. uid, err := res.GetString("uid")
  119. if err != nil {
  120. log.Warnf("CostModel.ComputeAllocation: UID ingestion enabled, but query result missing field: %s", err)
  121. } else {
  122. newKey := newPodKey(cluster, namespace, podName+" "+uid)
  123. podUIDKeyMap[key] = append(podUIDKeyMap[key], newKey)
  124. key = newKey
  125. }
  126. }
  127. allocStart, allocEnd := calculateStartEndFromIsRunning(res, resolution, window)
  128. if allocStart.IsZero() || allocEnd.IsZero() {
  129. continue
  130. }
  131. // Set start if unset or this datum's start time is earlier than the
  132. // current earliest time.
  133. if _, ok := clusterStart[cluster]; !ok || allocStart.Before(clusterStart[cluster]) {
  134. clusterStart[cluster] = allocStart
  135. }
  136. // Set end if unset or this datum's end time is later than the
  137. // current latest time.
  138. if _, ok := clusterEnd[cluster]; !ok || allocEnd.After(clusterEnd[cluster]) {
  139. clusterEnd[cluster] = allocEnd
  140. }
  141. if thisPod, ok := podMap[key]; ok {
  142. // Pod has already been recorded, so update it accordingly
  143. if allocStart.Before(thisPod.Start) {
  144. thisPod.Start = allocStart
  145. }
  146. if allocEnd.After(thisPod.End) {
  147. thisPod.End = allocEnd
  148. }
  149. } else {
  150. // pod has not been recorded yet, so insert it
  151. podMap[key] = &pod{
  152. Window: window.Clone(),
  153. Start: allocStart,
  154. End: allocEnd,
  155. Key: key,
  156. Allocations: map[string]*kubecost.Allocation{},
  157. }
  158. }
  159. }
  160. }
  161. func applyCPUCoresAllocated(podMap map[podKey]*pod, resCPUCoresAllocated []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey) {
  162. for _, res := range resCPUCoresAllocated {
  163. key, err := resultPodKey(res, env.GetPromClusterLabel(), "namespace")
  164. if err != nil {
  165. log.DedupedWarningf(10, "CostModel.ComputeAllocation: CPU allocation result missing field: %s", err)
  166. continue
  167. }
  168. container, err := res.GetString("container")
  169. if err != nil {
  170. log.DedupedWarningf(10, "CostModel.ComputeAllocation: CPU allocation query result missing 'container': %s", key)
  171. continue
  172. }
  173. var pods []*pod
  174. if thisPod, ok := podMap[key]; !ok {
  175. if uidKeys, ok := podUIDKeyMap[key]; ok {
  176. for _, uidKey := range uidKeys {
  177. thisPod, ok = podMap[uidKey]
  178. if ok {
  179. pods = append(pods, thisPod)
  180. }
  181. }
  182. } else {
  183. continue
  184. }
  185. } else {
  186. pods = []*pod{thisPod}
  187. }
  188. for _, thisPod := range pods {
  189. if _, ok := thisPod.Allocations[container]; !ok {
  190. thisPod.appendContainer(container)
  191. }
  192. cpuCores := res.Values[0].Value
  193. if cpuCores > MAX_CPU_CAP {
  194. log.Infof("[WARNING] Very large cpu allocation, clamping to %f", res.Values[0].Value*(thisPod.Allocations[container].Minutes()/60.0))
  195. cpuCores = 0.0
  196. }
  197. hours := thisPod.Allocations[container].Minutes() / 60.0
  198. thisPod.Allocations[container].CPUCoreHours = cpuCores * hours
  199. node, err := res.GetString("node")
  200. if err != nil {
  201. log.Warnf("CostModel.ComputeAllocation: CPU allocation query result missing 'node': %s", key)
  202. continue
  203. }
  204. thisPod.Allocations[container].Properties.Node = node
  205. thisPod.Node = node
  206. }
  207. }
  208. }
  209. func applyCPUCoresRequested(podMap map[podKey]*pod, resCPUCoresRequested []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey) {
  210. for _, res := range resCPUCoresRequested {
  211. key, err := resultPodKey(res, env.GetPromClusterLabel(), "namespace")
  212. if err != nil {
  213. log.DedupedWarningf(10, "CostModel.ComputeAllocation: CPU request result missing field: %s", err)
  214. continue
  215. }
  216. container, err := res.GetString("container")
  217. if err != nil {
  218. log.DedupedWarningf(10, "CostModel.ComputeAllocation: CPU request query result missing 'container': %s", key)
  219. continue
  220. }
  221. var pods []*pod
  222. if thisPod, ok := podMap[key]; !ok {
  223. if uidKeys, ok := podUIDKeyMap[key]; ok {
  224. for _, uidKey := range uidKeys {
  225. thisPod, ok = podMap[uidKey]
  226. if ok {
  227. pods = append(pods, thisPod)
  228. }
  229. }
  230. } else {
  231. continue
  232. }
  233. } else {
  234. pods = []*pod{thisPod}
  235. }
  236. for _, thisPod := range pods {
  237. if _, ok := thisPod.Allocations[container]; !ok {
  238. thisPod.appendContainer(container)
  239. }
  240. thisPod.Allocations[container].CPUCoreRequestAverage = res.Values[0].Value
  241. // If CPU allocation is less than requests, set CPUCoreHours to
  242. // request level.
  243. if thisPod.Allocations[container].CPUCores() < res.Values[0].Value {
  244. thisPod.Allocations[container].CPUCoreHours = res.Values[0].Value * (thisPod.Allocations[container].Minutes() / 60.0)
  245. }
  246. if thisPod.Allocations[container].CPUCores() > MAX_CPU_CAP {
  247. log.Infof("[WARNING] Very large cpu allocation, clamping! to %f", res.Values[0].Value*(thisPod.Allocations[container].Minutes()/60.0))
  248. thisPod.Allocations[container].CPUCoreHours = res.Values[0].Value * (thisPod.Allocations[container].Minutes() / 60.0)
  249. }
  250. node, err := res.GetString("node")
  251. if err != nil {
  252. log.Warnf("CostModel.ComputeAllocation: CPU request query result missing 'node': %s", key)
  253. continue
  254. }
  255. thisPod.Allocations[container].Properties.Node = node
  256. thisPod.Node = node
  257. }
  258. }
  259. }
  260. func applyCPUCoresUsedAvg(podMap map[podKey]*pod, resCPUCoresUsedAvg []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey) {
  261. for _, res := range resCPUCoresUsedAvg {
  262. key, err := resultPodKey(res, env.GetPromClusterLabel(), "namespace")
  263. if err != nil {
  264. log.DedupedWarningf(10, "CostModel.ComputeAllocation: CPU usage avg result missing field: %s", err)
  265. continue
  266. }
  267. container, err := res.GetString("container")
  268. if container == "" || err != nil {
  269. container, err = res.GetString("container_name")
  270. if err != nil {
  271. log.DedupedWarningf(10, "CostModel.ComputeAllocation: CPU usage avg query result missing 'container': %s", key)
  272. continue
  273. }
  274. }
  275. var pods []*pod
  276. if thisPod, ok := podMap[key]; !ok {
  277. if uidKeys, ok := podUIDKeyMap[key]; ok {
  278. for _, uidKey := range uidKeys {
  279. thisPod, ok = podMap[uidKey]
  280. if ok {
  281. pods = append(pods, thisPod)
  282. }
  283. }
  284. } else {
  285. continue
  286. }
  287. } else {
  288. pods = []*pod{thisPod}
  289. }
  290. for _, thisPod := range pods {
  291. if _, ok := thisPod.Allocations[container]; !ok {
  292. thisPod.appendContainer(container)
  293. }
  294. thisPod.Allocations[container].CPUCoreUsageAverage = res.Values[0].Value
  295. if res.Values[0].Value > MAX_CPU_CAP {
  296. log.Infof("[WARNING] Very large cpu USAGE, dropping outlier")
  297. thisPod.Allocations[container].CPUCoreUsageAverage = 0.0
  298. }
  299. }
  300. }
  301. }
  302. func applyCPUCoresUsedMax(podMap map[podKey]*pod, resCPUCoresUsedMax []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey) {
  303. for _, res := range resCPUCoresUsedMax {
  304. key, err := resultPodKey(res, env.GetPromClusterLabel(), "namespace")
  305. if err != nil {
  306. log.DedupedWarningf(10, "CostModel.ComputeAllocation: CPU usage max result missing field: %s", err)
  307. continue
  308. }
  309. container, err := res.GetString("container")
  310. if container == "" || err != nil {
  311. container, err = res.GetString("container_name")
  312. if err != nil {
  313. log.DedupedWarningf(10, "CostModel.ComputeAllocation: CPU usage max query result missing 'container': %s", key)
  314. continue
  315. }
  316. }
  317. var pods []*pod
  318. if thisPod, ok := podMap[key]; !ok {
  319. if uidKeys, ok := podUIDKeyMap[key]; ok {
  320. for _, uidKey := range uidKeys {
  321. thisPod, ok = podMap[uidKey]
  322. if ok {
  323. pods = append(pods, thisPod)
  324. }
  325. }
  326. } else {
  327. continue
  328. }
  329. } else {
  330. pods = []*pod{thisPod}
  331. }
  332. for _, thisPod := range pods {
  333. if _, ok := thisPod.Allocations[container]; !ok {
  334. thisPod.appendContainer(container)
  335. }
  336. if thisPod.Allocations[container].RawAllocationOnly == nil {
  337. thisPod.Allocations[container].RawAllocationOnly = &kubecost.RawAllocationOnlyData{
  338. CPUCoreUsageMax: res.Values[0].Value,
  339. }
  340. } else {
  341. thisPod.Allocations[container].RawAllocationOnly.CPUCoreUsageMax = res.Values[0].Value
  342. }
  343. }
  344. }
  345. }
  346. func applyRAMBytesAllocated(podMap map[podKey]*pod, resRAMBytesAllocated []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey) {
  347. for _, res := range resRAMBytesAllocated {
  348. key, err := resultPodKey(res, env.GetPromClusterLabel(), "namespace")
  349. if err != nil {
  350. log.DedupedWarningf(10, "CostModel.ComputeAllocation: RAM allocation result missing field: %s", err)
  351. continue
  352. }
  353. container, err := res.GetString("container")
  354. if err != nil {
  355. log.DedupedWarningf(10, "CostModel.ComputeAllocation: RAM allocation query result missing 'container': %s", key)
  356. continue
  357. }
  358. var pods []*pod
  359. if thisPod, ok := podMap[key]; !ok {
  360. if uidKeys, ok := podUIDKeyMap[key]; ok {
  361. for _, uidKey := range uidKeys {
  362. thisPod, ok = podMap[uidKey]
  363. if ok {
  364. pods = append(pods, thisPod)
  365. }
  366. }
  367. } else {
  368. continue
  369. }
  370. } else {
  371. pods = []*pod{thisPod}
  372. }
  373. for _, thisPod := range pods {
  374. if _, ok := thisPod.Allocations[container]; !ok {
  375. thisPod.appendContainer(container)
  376. }
  377. ramBytes := res.Values[0].Value
  378. hours := thisPod.Allocations[container].Minutes() / 60.0
  379. thisPod.Allocations[container].RAMByteHours = ramBytes * hours
  380. node, err := res.GetString("node")
  381. if err != nil {
  382. log.Warnf("CostModel.ComputeAllocation: RAM allocation query result missing 'node': %s", key)
  383. continue
  384. }
  385. thisPod.Allocations[container].Properties.Node = node
  386. thisPod.Node = node
  387. }
  388. }
  389. }
  390. func applyRAMBytesRequested(podMap map[podKey]*pod, resRAMBytesRequested []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey) {
  391. for _, res := range resRAMBytesRequested {
  392. key, err := resultPodKey(res, env.GetPromClusterLabel(), "namespace")
  393. if err != nil {
  394. log.DedupedWarningf(10, "CostModel.ComputeAllocation: RAM request result missing field: %s", err)
  395. continue
  396. }
  397. container, err := res.GetString("container")
  398. if err != nil {
  399. log.DedupedWarningf(10, "CostModel.ComputeAllocation: RAM request query result missing 'container': %s", key)
  400. continue
  401. }
  402. var pods []*pod
  403. if thisPod, ok := podMap[key]; !ok {
  404. if uidKeys, ok := podUIDKeyMap[key]; ok {
  405. for _, uidKey := range uidKeys {
  406. thisPod, ok = podMap[uidKey]
  407. if ok {
  408. pods = append(pods, thisPod)
  409. }
  410. }
  411. } else {
  412. continue
  413. }
  414. } else {
  415. pods = []*pod{thisPod}
  416. }
  417. for _, pod := range pods {
  418. if _, ok := pod.Allocations[container]; !ok {
  419. pod.appendContainer(container)
  420. }
  421. pod.Allocations[container].RAMBytesRequestAverage = res.Values[0].Value
  422. // If RAM allocation is less than requests, set RAMByteHours to
  423. // request level.
  424. if pod.Allocations[container].RAMBytes() < res.Values[0].Value {
  425. pod.Allocations[container].RAMByteHours = res.Values[0].Value * (pod.Allocations[container].Minutes() / 60.0)
  426. }
  427. node, err := res.GetString("node")
  428. if err != nil {
  429. log.Warnf("CostModel.ComputeAllocation: RAM request query result missing 'node': %s", key)
  430. continue
  431. }
  432. pod.Allocations[container].Properties.Node = node
  433. pod.Node = node
  434. }
  435. }
  436. }
  437. func applyRAMBytesUsedAvg(podMap map[podKey]*pod, resRAMBytesUsedAvg []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey) {
  438. for _, res := range resRAMBytesUsedAvg {
  439. key, err := resultPodKey(res, env.GetPromClusterLabel(), "namespace")
  440. if err != nil {
  441. log.DedupedWarningf(10, "CostModel.ComputeAllocation: RAM avg usage result missing field: %s", err)
  442. continue
  443. }
  444. container, err := res.GetString("container")
  445. if container == "" || err != nil {
  446. container, err = res.GetString("container_name")
  447. if err != nil {
  448. log.DedupedWarningf(10, "CostModel.ComputeAllocation: RAM usage avg query result missing 'container': %s", key)
  449. continue
  450. }
  451. }
  452. var pods []*pod
  453. if thisPod, ok := podMap[key]; !ok {
  454. if uidKeys, ok := podUIDKeyMap[key]; ok {
  455. for _, uidKey := range uidKeys {
  456. thisPod, ok = podMap[uidKey]
  457. if ok {
  458. pods = append(pods, thisPod)
  459. }
  460. }
  461. } else {
  462. continue
  463. }
  464. } else {
  465. pods = []*pod{thisPod}
  466. }
  467. for _, thisPod := range pods {
  468. if _, ok := thisPod.Allocations[container]; !ok {
  469. thisPod.appendContainer(container)
  470. }
  471. thisPod.Allocations[container].RAMBytesUsageAverage = res.Values[0].Value
  472. }
  473. }
  474. }
  475. func applyRAMBytesUsedMax(podMap map[podKey]*pod, resRAMBytesUsedMax []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey) {
  476. for _, res := range resRAMBytesUsedMax {
  477. key, err := resultPodKey(res, env.GetPromClusterLabel(), "namespace")
  478. if err != nil {
  479. log.DedupedWarningf(10, "CostModel.ComputeAllocation: RAM usage max result missing field: %s", err)
  480. continue
  481. }
  482. container, err := res.GetString("container")
  483. if container == "" || err != nil {
  484. container, err = res.GetString("container_name")
  485. if err != nil {
  486. log.DedupedWarningf(10, "CostModel.ComputeAllocation: RAM usage max query result missing 'container': %s", key)
  487. continue
  488. }
  489. }
  490. var pods []*pod
  491. if thisPod, ok := podMap[key]; !ok {
  492. if uidKeys, ok := podUIDKeyMap[key]; ok {
  493. for _, uidKey := range uidKeys {
  494. thisPod, ok = podMap[uidKey]
  495. if ok {
  496. pods = append(pods, thisPod)
  497. }
  498. }
  499. } else {
  500. continue
  501. }
  502. } else {
  503. pods = []*pod{thisPod}
  504. }
  505. for _, thisPod := range pods {
  506. if _, ok := thisPod.Allocations[container]; !ok {
  507. thisPod.appendContainer(container)
  508. }
  509. if thisPod.Allocations[container].RawAllocationOnly == nil {
  510. thisPod.Allocations[container].RawAllocationOnly = &kubecost.RawAllocationOnlyData{
  511. RAMBytesUsageMax: res.Values[0].Value,
  512. }
  513. } else {
  514. thisPod.Allocations[container].RawAllocationOnly.RAMBytesUsageMax = res.Values[0].Value
  515. }
  516. }
  517. }
  518. }
  519. func applyGPUsAllocated(podMap map[podKey]*pod, resGPUsRequested []*prom.QueryResult, resGPUsAllocated []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey) {
  520. if len(resGPUsAllocated) > 0 { // Use the new query, when it's become available in a window
  521. resGPUsRequested = resGPUsAllocated
  522. }
  523. for _, res := range resGPUsRequested {
  524. key, err := resultPodKey(res, env.GetPromClusterLabel(), "namespace")
  525. if err != nil {
  526. log.DedupedWarningf(10, "CostModel.ComputeAllocation: GPU request result missing field: %s", err)
  527. continue
  528. }
  529. container, err := res.GetString("container")
  530. if err != nil {
  531. log.DedupedWarningf(10, "CostModel.ComputeAllocation: GPU request query result missing 'container': %s", key)
  532. continue
  533. }
  534. var pods []*pod
  535. if thisPod, ok := podMap[key]; !ok {
  536. if uidKeys, ok := podUIDKeyMap[key]; ok {
  537. for _, uidKey := range uidKeys {
  538. thisPod, ok = podMap[uidKey]
  539. if ok {
  540. pods = append(pods, thisPod)
  541. }
  542. }
  543. } else {
  544. continue
  545. }
  546. } else {
  547. pods = []*pod{thisPod}
  548. }
  549. for _, thisPod := range pods {
  550. if _, ok := thisPod.Allocations[container]; !ok {
  551. thisPod.appendContainer(container)
  552. }
  553. hrs := thisPod.Allocations[container].Minutes() / 60.0
  554. thisPod.Allocations[container].GPUHours = res.Values[0].Value * hrs
  555. }
  556. }
  557. }
  558. func applyNetworkTotals(podMap map[podKey]*pod, resNetworkTransferBytes []*prom.QueryResult, resNetworkReceiveBytes []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey) {
  559. for _, res := range resNetworkTransferBytes {
  560. podKey, err := resultPodKey(res, env.GetPromClusterLabel(), "namespace")
  561. if err != nil {
  562. log.DedupedWarningf(10, "CostModel.ComputeAllocation: Network Transfer Bytes query result missing field: %s", err)
  563. continue
  564. }
  565. var pods []*pod
  566. if thisPod, ok := podMap[podKey]; !ok {
  567. if uidKeys, ok := podUIDKeyMap[podKey]; ok {
  568. for _, uidKey := range uidKeys {
  569. thisPod, ok = podMap[uidKey]
  570. if ok {
  571. pods = append(pods, thisPod)
  572. }
  573. }
  574. } else {
  575. continue
  576. }
  577. } else {
  578. pods = []*pod{thisPod}
  579. }
  580. for _, thisPod := range pods {
  581. for _, alloc := range thisPod.Allocations {
  582. alloc.NetworkTransferBytes = res.Values[0].Value / float64(len(thisPod.Allocations)) / float64(len(pods))
  583. }
  584. }
  585. }
  586. for _, res := range resNetworkReceiveBytes {
  587. podKey, err := resultPodKey(res, env.GetPromClusterLabel(), "namespace")
  588. if err != nil {
  589. log.DedupedWarningf(10, "CostModel.ComputeAllocation: Network Receive Bytes query result missing field: %s", err)
  590. continue
  591. }
  592. var pods []*pod
  593. if thisPod, ok := podMap[podKey]; !ok {
  594. if uidKeys, ok := podUIDKeyMap[podKey]; ok {
  595. for _, uidKey := range uidKeys {
  596. thisPod, ok = podMap[uidKey]
  597. if ok {
  598. pods = append(pods, thisPod)
  599. }
  600. }
  601. } else {
  602. continue
  603. }
  604. } else {
  605. pods = []*pod{thisPod}
  606. }
  607. for _, thisPod := range pods {
  608. for _, alloc := range thisPod.Allocations {
  609. alloc.NetworkReceiveBytes = res.Values[0].Value / float64(len(thisPod.Allocations)) / float64(len(pods))
  610. }
  611. }
  612. }
  613. }
  614. func applyNetworkAllocation(podMap map[podKey]*pod, resNetworkGiB []*prom.QueryResult, resNetworkCostPerGiB []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey, networkCostSubType string) {
  615. costPerGiBByCluster := map[string]float64{}
  616. for _, res := range resNetworkCostPerGiB {
  617. cluster, err := res.GetString(env.GetPromClusterLabel())
  618. if err != nil {
  619. cluster = env.GetClusterID()
  620. }
  621. costPerGiBByCluster[cluster] = res.Values[0].Value
  622. }
  623. for _, res := range resNetworkGiB {
  624. podKey, err := resultPodKey(res, env.GetPromClusterLabel(), "namespace")
  625. if err != nil {
  626. log.DedupedWarningf(10, "CostModel.ComputeAllocation: Network allocation query result missing field: %s", err)
  627. continue
  628. }
  629. var pods []*pod
  630. if thisPod, ok := podMap[podKey]; !ok {
  631. if uidKeys, ok := podUIDKeyMap[podKey]; ok {
  632. for _, uidKey := range uidKeys {
  633. thisPod, ok = podMap[uidKey]
  634. if ok {
  635. pods = append(pods, thisPod)
  636. }
  637. }
  638. } else {
  639. continue
  640. }
  641. } else {
  642. pods = []*pod{thisPod}
  643. }
  644. for _, thisPod := range pods {
  645. for _, alloc := range thisPod.Allocations {
  646. gib := res.Values[0].Value / float64(len(thisPod.Allocations))
  647. costPerGiB := costPerGiBByCluster[podKey.Cluster]
  648. currentNetworkSubCost := gib * costPerGiB / float64(len(pods))
  649. switch networkCostSubType {
  650. case networkCrossZoneCost:
  651. alloc.NetworkCrossZoneCost = currentNetworkSubCost
  652. case networkCrossRegionCost:
  653. alloc.NetworkCrossRegionCost = currentNetworkSubCost
  654. case networkInternetCost:
  655. alloc.NetworkInternetCost = currentNetworkSubCost
  656. default:
  657. log.Warnf("CostModel.applyNetworkAllocation: unknown network subtype passed to the function: %s", networkCostSubType)
  658. }
  659. alloc.NetworkCost += currentNetworkSubCost
  660. }
  661. }
  662. }
  663. }
  664. func resToNodeLabels(resNodeLabels []*prom.QueryResult) map[nodeKey]map[string]string {
  665. nodeLabels := map[nodeKey]map[string]string{}
  666. for _, res := range resNodeLabels {
  667. nodeKey, err := resultNodeKey(res, env.GetPromClusterLabel(), "node")
  668. if err != nil {
  669. continue
  670. }
  671. if _, ok := nodeLabels[nodeKey]; !ok {
  672. nodeLabels[nodeKey] = map[string]string{}
  673. }
  674. for _, rawK := range env.GetAllocationNodeLabelsIncludeList() {
  675. labels := res.GetLabels()
  676. // Sanitize the given label name to match Prometheus formatting
  677. // e.g. topology.kubernetes.io/zone => topology_kubernetes_io_zone
  678. k := prom.SanitizeLabelName(rawK)
  679. if v, ok := labels[k]; ok {
  680. nodeLabels[nodeKey][k] = v
  681. continue
  682. }
  683. // Try with the "label_" prefix, if not found
  684. // e.g. topology_kubernetes_io_zone => label_topology_kubernetes_io_zone
  685. k = fmt.Sprintf("label_%s", k)
  686. if v, ok := labels[k]; ok {
  687. nodeLabels[nodeKey][k] = v
  688. }
  689. }
  690. }
  691. return nodeLabels
  692. }
  693. func resToNamespaceLabels(resNamespaceLabels []*prom.QueryResult) map[namespaceKey]map[string]string {
  694. namespaceLabels := map[namespaceKey]map[string]string{}
  695. for _, res := range resNamespaceLabels {
  696. nsKey, err := resultNamespaceKey(res, env.GetPromClusterLabel(), "namespace")
  697. if err != nil {
  698. continue
  699. }
  700. if _, ok := namespaceLabels[nsKey]; !ok {
  701. namespaceLabels[nsKey] = map[string]string{}
  702. }
  703. for k, l := range res.GetLabels() {
  704. namespaceLabels[nsKey][k] = l
  705. }
  706. }
  707. return namespaceLabels
  708. }
  709. func resToPodLabels(resPodLabels []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey, ingestPodUID bool) map[podKey]map[string]string {
  710. podLabels := map[podKey]map[string]string{}
  711. for _, res := range resPodLabels {
  712. key, err := resultPodKey(res, env.GetPromClusterLabel(), "namespace")
  713. if err != nil {
  714. continue
  715. }
  716. var keys []podKey
  717. if ingestPodUID {
  718. if uidKeys, ok := podUIDKeyMap[key]; ok {
  719. keys = append(keys, uidKeys...)
  720. }
  721. } else {
  722. keys = []podKey{key}
  723. }
  724. for _, key := range keys {
  725. if _, ok := podLabels[key]; !ok {
  726. podLabels[key] = map[string]string{}
  727. }
  728. for k, l := range res.GetLabels() {
  729. podLabels[key][k] = l
  730. }
  731. }
  732. }
  733. return podLabels
  734. }
  735. func resToNamespaceAnnotations(resNamespaceAnnotations []*prom.QueryResult) map[string]map[string]string {
  736. namespaceAnnotations := map[string]map[string]string{}
  737. for _, res := range resNamespaceAnnotations {
  738. namespace, err := res.GetString("namespace")
  739. if err != nil {
  740. continue
  741. }
  742. if _, ok := namespaceAnnotations[namespace]; !ok {
  743. namespaceAnnotations[namespace] = map[string]string{}
  744. }
  745. for k, l := range res.GetAnnotations() {
  746. namespaceAnnotations[namespace][k] = l
  747. }
  748. }
  749. return namespaceAnnotations
  750. }
  751. func resToPodAnnotations(resPodAnnotations []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey, ingestPodUID bool) map[podKey]map[string]string {
  752. podAnnotations := map[podKey]map[string]string{}
  753. for _, res := range resPodAnnotations {
  754. key, err := resultPodKey(res, env.GetPromClusterLabel(), "namespace")
  755. if err != nil {
  756. continue
  757. }
  758. var keys []podKey
  759. if ingestPodUID {
  760. if uidKeys, ok := podUIDKeyMap[key]; ok {
  761. keys = append(keys, uidKeys...)
  762. }
  763. } else {
  764. keys = []podKey{key}
  765. }
  766. for _, key := range keys {
  767. if _, ok := podAnnotations[key]; !ok {
  768. podAnnotations[key] = map[string]string{}
  769. }
  770. for k, l := range res.GetAnnotations() {
  771. podAnnotations[key][k] = l
  772. }
  773. }
  774. }
  775. return podAnnotations
  776. }
  777. func applyLabels(podMap map[podKey]*pod, nodeLabels map[nodeKey]map[string]string, namespaceLabels map[namespaceKey]map[string]string, podLabels map[podKey]map[string]string) {
  778. for podKey, pod := range podMap {
  779. for _, alloc := range pod.Allocations {
  780. allocLabels := alloc.Properties.Labels
  781. if allocLabels == nil {
  782. allocLabels = make(map[string]string)
  783. }
  784. nsLabels := alloc.Properties.NamespaceLabels
  785. if nsLabels == nil {
  786. nsLabels = make(map[string]string)
  787. }
  788. // Apply node labels first, then namespace labels, then pod labels
  789. // so that pod labels overwrite namespace labels, which overwrite
  790. // node labels.
  791. if nodeLabels != nil {
  792. nodeKey := newNodeKey(pod.Key.Cluster, pod.Node)
  793. if labels, ok := nodeLabels[nodeKey]; ok {
  794. for k, v := range labels {
  795. allocLabels[k] = v
  796. }
  797. }
  798. }
  799. nsKey := podKey.namespaceKey
  800. if labels, ok := namespaceLabels[nsKey]; ok {
  801. for k, v := range labels {
  802. allocLabels[k] = v
  803. nsLabels[k] = v
  804. }
  805. }
  806. if labels, ok := podLabels[podKey]; ok {
  807. for k, v := range labels {
  808. allocLabels[k] = v
  809. }
  810. }
  811. alloc.Properties.Labels = allocLabels
  812. alloc.Properties.NamespaceLabels = nsLabels
  813. }
  814. }
  815. }
  816. func applyAnnotations(podMap map[podKey]*pod, namespaceAnnotations map[string]map[string]string, podAnnotations map[podKey]map[string]string) {
  817. for key, pod := range podMap {
  818. for _, alloc := range pod.Allocations {
  819. allocAnnotations := alloc.Properties.Annotations
  820. if allocAnnotations == nil {
  821. allocAnnotations = make(map[string]string)
  822. }
  823. nsAnnotations := alloc.Properties.NamespaceAnnotations
  824. if nsAnnotations == nil {
  825. nsAnnotations = make(map[string]string)
  826. }
  827. // Apply namespace annotations first, then pod annotations so that
  828. // pod labels overwrite namespace labels.
  829. if labels, ok := namespaceAnnotations[key.Namespace]; ok {
  830. for k, v := range labels {
  831. allocAnnotations[k] = v
  832. nsAnnotations[k] = v
  833. }
  834. }
  835. if labels, ok := podAnnotations[key]; ok {
  836. for k, v := range labels {
  837. allocAnnotations[k] = v
  838. }
  839. }
  840. alloc.Properties.Annotations = allocAnnotations
  841. alloc.Properties.NamespaceAnnotations = nsAnnotations
  842. }
  843. }
  844. }
  845. func resToDeploymentLabels(resDeploymentLabels []*prom.QueryResult) map[controllerKey]map[string]string {
  846. deploymentLabels := map[controllerKey]map[string]string{}
  847. for _, res := range resDeploymentLabels {
  848. controllerKey, err := resultDeploymentKey(res, env.GetPromClusterLabel(), "namespace", "deployment")
  849. if err != nil {
  850. continue
  851. }
  852. if _, ok := deploymentLabels[controllerKey]; !ok {
  853. deploymentLabels[controllerKey] = map[string]string{}
  854. }
  855. for k, l := range res.GetLabels() {
  856. deploymentLabels[controllerKey][k] = l
  857. }
  858. }
  859. // Prune duplicate deployments. That is, if the same deployment exists with
  860. // hyphens instead of underscores, keep the one that uses hyphens.
  861. for key := range deploymentLabels {
  862. if strings.Contains(key.Controller, "_") {
  863. duplicateController := strings.Replace(key.Controller, "_", "-", -1)
  864. duplicateKey := newControllerKey(key.Cluster, key.Namespace, key.ControllerKind, duplicateController)
  865. if _, ok := deploymentLabels[duplicateKey]; ok {
  866. delete(deploymentLabels, key)
  867. }
  868. }
  869. }
  870. return deploymentLabels
  871. }
  872. func resToStatefulSetLabels(resStatefulSetLabels []*prom.QueryResult) map[controllerKey]map[string]string {
  873. statefulSetLabels := map[controllerKey]map[string]string{}
  874. for _, res := range resStatefulSetLabels {
  875. controllerKey, err := resultStatefulSetKey(res, env.GetPromClusterLabel(), "namespace", "statefulSet")
  876. if err != nil {
  877. continue
  878. }
  879. if _, ok := statefulSetLabels[controllerKey]; !ok {
  880. statefulSetLabels[controllerKey] = map[string]string{}
  881. }
  882. for k, l := range res.GetLabels() {
  883. statefulSetLabels[controllerKey][k] = l
  884. }
  885. }
  886. // Prune duplicate stateful sets. That is, if the same stateful set exists
  887. // with hyphens instead of underscores, keep the one that uses hyphens.
  888. for key := range statefulSetLabels {
  889. if strings.Contains(key.Controller, "_") {
  890. duplicateController := strings.Replace(key.Controller, "_", "-", -1)
  891. duplicateKey := newControllerKey(key.Cluster, key.Namespace, key.ControllerKind, duplicateController)
  892. if _, ok := statefulSetLabels[duplicateKey]; ok {
  893. delete(statefulSetLabels, key)
  894. }
  895. }
  896. }
  897. return statefulSetLabels
  898. }
  899. func labelsToPodControllerMap(podLabels map[podKey]map[string]string, controllerLabels map[controllerKey]map[string]string) map[podKey]controllerKey {
  900. podControllerMap := map[podKey]controllerKey{}
  901. // For each controller, turn the labels into a selector and attempt to
  902. // match it with each set of pod labels. A match indicates that the pod
  903. // belongs to the controller.
  904. for cKey, cLabels := range controllerLabels {
  905. selector := labels.Set(cLabels).AsSelectorPreValidated()
  906. for pKey, pLabels := range podLabels {
  907. // If the pod is in a different cluster or namespace, there is
  908. // no need to compare the labels.
  909. if cKey.Cluster != pKey.Cluster || cKey.Namespace != pKey.Namespace {
  910. continue
  911. }
  912. podLabelSet := labels.Set(pLabels)
  913. if selector.Matches(podLabelSet) {
  914. if _, ok := podControllerMap[pKey]; ok {
  915. log.DedupedWarningf(5, "CostModel.ComputeAllocation: PodControllerMap match already exists: %s matches %s and %s", pKey, podControllerMap[pKey], cKey)
  916. }
  917. podControllerMap[pKey] = cKey
  918. }
  919. }
  920. }
  921. return podControllerMap
  922. }
  923. func resToPodDaemonSetMap(resDaemonSetLabels []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey, ingestPodUID bool) map[podKey]controllerKey {
  924. daemonSetLabels := map[podKey]controllerKey{}
  925. for _, res := range resDaemonSetLabels {
  926. controllerKey, err := resultDaemonSetKey(res, env.GetPromClusterLabel(), "namespace", "owner_name")
  927. if err != nil {
  928. continue
  929. }
  930. pod, err := res.GetString("pod")
  931. if err != nil {
  932. log.Warnf("CostModel.ComputeAllocation: DaemonSetLabel result without pod: %s", controllerKey)
  933. }
  934. key := newPodKey(controllerKey.Cluster, controllerKey.Namespace, pod)
  935. var keys []podKey
  936. if ingestPodUID {
  937. if uidKeys, ok := podUIDKeyMap[key]; ok {
  938. keys = append(keys, uidKeys...)
  939. }
  940. } else {
  941. keys = []podKey{key}
  942. }
  943. for _, key := range keys {
  944. daemonSetLabels[key] = controllerKey
  945. }
  946. }
  947. return daemonSetLabels
  948. }
  949. func resToPodJobMap(resJobLabels []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey, ingestPodUID bool) map[podKey]controllerKey {
  950. jobLabels := map[podKey]controllerKey{}
  951. for _, res := range resJobLabels {
  952. controllerKey, err := resultJobKey(res, env.GetPromClusterLabel(), "namespace", "owner_name")
  953. if err != nil {
  954. continue
  955. }
  956. // Convert the name of Jobs generated by CronJobs to the name of the
  957. // CronJob by stripping the timestamp off the end.
  958. match := isCron.FindStringSubmatch(controllerKey.Controller)
  959. if match != nil {
  960. controllerKey.Controller = match[1]
  961. }
  962. pod, err := res.GetString("pod")
  963. if err != nil {
  964. log.Warnf("CostModel.ComputeAllocation: JobLabel result without pod: %s", controllerKey)
  965. }
  966. key := newPodKey(controllerKey.Cluster, controllerKey.Namespace, pod)
  967. var keys []podKey
  968. if ingestPodUID {
  969. if uidKeys, ok := podUIDKeyMap[key]; ok {
  970. keys = append(keys, uidKeys...)
  971. }
  972. } else {
  973. keys = []podKey{key}
  974. }
  975. for _, key := range keys {
  976. jobLabels[key] = controllerKey
  977. }
  978. }
  979. return jobLabels
  980. }
  981. func resToPodReplicaSetMap(resPodsWithReplicaSetOwner []*prom.QueryResult, resReplicaSetsWithoutOwners []*prom.QueryResult, resReplicaSetsWithRolloutOwner []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey, ingestPodUID bool) map[podKey]controllerKey {
  982. // Build out set of ReplicaSets that have no owners, themselves, such that
  983. // the ReplicaSet should be used as the owner of the Pods it controls.
  984. // (This should exclude, for example, ReplicaSets that are controlled by
  985. // Deployments, in which case the Deployment should be the pod's owner.)
  986. // Additionally, add to this set of ReplicaSets those ReplicaSets that
  987. // are owned by a Rollout
  988. replicaSets := map[controllerKey]struct{}{}
  989. // Create unowned ReplicaSet controller keys
  990. for _, res := range resReplicaSetsWithoutOwners {
  991. controllerKey, err := resultReplicaSetKey(res, env.GetPromClusterLabel(), "namespace", "replicaset")
  992. if err != nil {
  993. continue
  994. }
  995. replicaSets[controllerKey] = struct{}{}
  996. }
  997. // Create Rollout-owned ReplicaSet controller keys
  998. for _, res := range resReplicaSetsWithRolloutOwner {
  999. controllerKey, err := resultReplicaSetRolloutKey(res, env.GetPromClusterLabel(), "namespace", "replicaset")
  1000. if err != nil {
  1001. continue
  1002. }
  1003. replicaSets[controllerKey] = struct{}{}
  1004. }
  1005. // Create the mapping of Pods to ReplicaSets, ignoring any ReplicaSets that
  1006. // do not appear in the set of unowned/Rollout-owned ReplicaSets above.
  1007. podToReplicaSet := map[podKey]controllerKey{}
  1008. for _, res := range resPodsWithReplicaSetOwner {
  1009. // First, check if this pod is owned by an unowned ReplicaSet
  1010. controllerKey, err := resultReplicaSetKey(res, env.GetPromClusterLabel(), "namespace", "owner_name")
  1011. if err != nil {
  1012. continue
  1013. } else if _, ok := replicaSets[controllerKey]; !ok {
  1014. // If the pod is not owned by an unowned ReplicaSet, check if
  1015. // it's owned by a Rollout-owned ReplicaSet
  1016. controllerKey, err = resultReplicaSetRolloutKey(res, env.GetPromClusterLabel(), "namespace", "owner_name")
  1017. if err != nil {
  1018. continue
  1019. } else if _, ok := replicaSets[controllerKey]; !ok {
  1020. continue
  1021. }
  1022. }
  1023. pod, err := res.GetString("pod")
  1024. if err != nil {
  1025. log.Warnf("CostModel.ComputeAllocation: ReplicaSet result without pod: %s", controllerKey)
  1026. }
  1027. key := newPodKey(controllerKey.Cluster, controllerKey.Namespace, pod)
  1028. var keys []podKey
  1029. if ingestPodUID {
  1030. if uidKeys, ok := podUIDKeyMap[key]; ok {
  1031. keys = append(keys, uidKeys...)
  1032. }
  1033. } else {
  1034. keys = []podKey{key}
  1035. }
  1036. for _, key := range keys {
  1037. podToReplicaSet[key] = controllerKey
  1038. }
  1039. }
  1040. return podToReplicaSet
  1041. }
  1042. func applyControllersToPods(podMap map[podKey]*pod, podControllerMap map[podKey]controllerKey) {
  1043. for key, pod := range podMap {
  1044. for _, alloc := range pod.Allocations {
  1045. if controllerKey, ok := podControllerMap[key]; ok {
  1046. alloc.Properties.ControllerKind = controllerKey.ControllerKind
  1047. alloc.Properties.Controller = controllerKey.Controller
  1048. }
  1049. }
  1050. }
  1051. }
  1052. /* Service Helpers */
  1053. func getServiceLabels(resServiceLabels []*prom.QueryResult) map[serviceKey]map[string]string {
  1054. serviceLabels := map[serviceKey]map[string]string{}
  1055. for _, res := range resServiceLabels {
  1056. serviceKey, err := resultServiceKey(res, env.GetPromClusterLabel(), "namespace", "service")
  1057. if err != nil {
  1058. continue
  1059. }
  1060. if _, ok := serviceLabels[serviceKey]; !ok {
  1061. serviceLabels[serviceKey] = map[string]string{}
  1062. }
  1063. for k, l := range res.GetLabels() {
  1064. serviceLabels[serviceKey][k] = l
  1065. }
  1066. }
  1067. // Prune duplicate services. That is, if the same service exists with
  1068. // hyphens instead of underscores, keep the one that uses hyphens.
  1069. for key := range serviceLabels {
  1070. if strings.Contains(key.Service, "_") {
  1071. duplicateService := strings.Replace(key.Service, "_", "-", -1)
  1072. duplicateKey := newServiceKey(key.Cluster, key.Namespace, duplicateService)
  1073. if _, ok := serviceLabels[duplicateKey]; ok {
  1074. delete(serviceLabels, key)
  1075. }
  1076. }
  1077. }
  1078. return serviceLabels
  1079. }
  1080. func applyServicesToPods(podMap map[podKey]*pod, podLabels map[podKey]map[string]string, allocsByService map[serviceKey][]*kubecost.Allocation, serviceLabels map[serviceKey]map[string]string) {
  1081. podServicesMap := map[podKey][]serviceKey{}
  1082. // For each service, turn the labels into a selector and attempt to
  1083. // match it with each set of pod labels. A match indicates that the pod
  1084. // belongs to the service.
  1085. for sKey, sLabels := range serviceLabels {
  1086. selector := labels.Set(sLabels).AsSelectorPreValidated()
  1087. for pKey, pLabels := range podLabels {
  1088. // If the pod is in a different cluster or namespace, there is
  1089. // no need to compare the labels.
  1090. if sKey.Cluster != pKey.Cluster || sKey.Namespace != pKey.Namespace {
  1091. continue
  1092. }
  1093. podLabelSet := labels.Set(pLabels)
  1094. if selector.Matches(podLabelSet) {
  1095. if _, ok := podServicesMap[pKey]; !ok {
  1096. podServicesMap[pKey] = []serviceKey{}
  1097. }
  1098. podServicesMap[pKey] = append(podServicesMap[pKey], sKey)
  1099. }
  1100. }
  1101. }
  1102. // For each allocation in each pod, attempt to find and apply the list of
  1103. // services associated with the allocation's pod.
  1104. for key, pod := range podMap {
  1105. for _, alloc := range pod.Allocations {
  1106. if sKeys, ok := podServicesMap[key]; ok {
  1107. services := []string{}
  1108. for _, sKey := range sKeys {
  1109. services = append(services, sKey.Service)
  1110. allocsByService[sKey] = append(allocsByService[sKey], alloc)
  1111. }
  1112. alloc.Properties.Services = services
  1113. }
  1114. }
  1115. }
  1116. }
  1117. func getLoadBalancerCosts(lbMap map[serviceKey]*lbCost, resLBCost, resLBActiveMins []*prom.QueryResult, resolution time.Duration) {
  1118. for _, res := range resLBActiveMins {
  1119. serviceKey, err := resultServiceKey(res, env.GetPromClusterLabel(), "namespace", "service_name")
  1120. if err != nil || len(res.Values) == 0 {
  1121. continue
  1122. }
  1123. lbStart, lbEnd := calculateStartAndEnd(res, resolution)
  1124. if lbStart.IsZero() || lbEnd.IsZero() {
  1125. log.Warnf("CostModel.ComputeAllocation: pvc %s has no running time", serviceKey)
  1126. }
  1127. lbMap[serviceKey] = &lbCost{
  1128. Start: lbStart,
  1129. End: lbEnd,
  1130. }
  1131. }
  1132. for _, res := range resLBCost {
  1133. serviceKey, err := resultServiceKey(res, env.GetPromClusterLabel(), "namespace", "service_name")
  1134. if err != nil {
  1135. continue
  1136. }
  1137. // Apply cost as price-per-hour * hours
  1138. if lb, ok := lbMap[serviceKey]; ok {
  1139. lbPricePerHr := res.Values[0].Value
  1140. hours := lb.End.Sub(lb.Start).Hours()
  1141. lb.TotalCost += lbPricePerHr * hours
  1142. } else {
  1143. log.DedupedWarningf(20, "CostModel: found minutes for key that does not exist: %s", serviceKey)
  1144. }
  1145. }
  1146. }
  1147. func applyLoadBalancersToPods(window kubecost.Window, podMap map[podKey]*pod, lbMap map[serviceKey]*lbCost, allocsByService map[serviceKey][]*kubecost.Allocation) {
  1148. for sKey, lb := range lbMap {
  1149. totalHours := 0.0
  1150. allocHours := make(map[*kubecost.Allocation]float64)
  1151. allocs, ok := allocsByService[sKey]
  1152. // if there are no allocations using the service, add its cost to the Unmounted pod for its cluster
  1153. if !ok {
  1154. pod := getUnmountedPodForCluster(window, podMap, sKey.Cluster)
  1155. pod.Allocations[kubecost.UnmountedSuffix].LoadBalancerCost += lb.TotalCost
  1156. pod.Allocations[kubecost.UnmountedSuffix].Properties.Services = append(pod.Allocations[kubecost.UnmountedSuffix].Properties.Services, sKey.Service)
  1157. }
  1158. // Add portion of load balancing cost to each allocation
  1159. // proportional to the total number of hours allocations used the load balancer
  1160. for _, alloc := range allocs {
  1161. // Determine the (start, end) of the relationship between the
  1162. // given lbCost and the associated Allocation so that a precise
  1163. // number of hours can be used to compute cumulative cost.
  1164. s, e := alloc.Start, alloc.End
  1165. if lb.Start.After(alloc.Start) {
  1166. s = lb.Start
  1167. }
  1168. if lb.End.Before(alloc.End) {
  1169. e = lb.End
  1170. }
  1171. hours := e.Sub(s).Hours()
  1172. // A negative number of hours signifies no overlap between the windows
  1173. if hours > 0 {
  1174. totalHours += hours
  1175. allocHours[alloc] = hours
  1176. }
  1177. }
  1178. // Distribute cost of service once total hours is calculated
  1179. for alloc, hours := range allocHours {
  1180. alloc.LoadBalancerCost += lb.TotalCost * hours / totalHours
  1181. }
  1182. // If there was no overlap apply to Unmounted pod
  1183. if len(allocHours) == 0 {
  1184. pod := getUnmountedPodForCluster(window, podMap, sKey.Cluster)
  1185. pod.Allocations[kubecost.UnmountedSuffix].LoadBalancerCost += lb.TotalCost
  1186. pod.Allocations[kubecost.UnmountedSuffix].Properties.Services = append(pod.Allocations[kubecost.UnmountedSuffix].Properties.Services, sKey.Service)
  1187. }
  1188. }
  1189. }
  1190. /* Node Helpers */
  1191. func applyNodeCostPerCPUHr(nodeMap map[nodeKey]*nodePricing, resNodeCostPerCPUHr []*prom.QueryResult) {
  1192. for _, res := range resNodeCostPerCPUHr {
  1193. cluster, err := res.GetString(env.GetPromClusterLabel())
  1194. if err != nil {
  1195. cluster = env.GetClusterID()
  1196. }
  1197. node, err := res.GetString("node")
  1198. if err != nil {
  1199. log.Warnf("CostModel.ComputeAllocation: Node CPU cost query result missing field: \"%s\" for node \"%s\"", err, node)
  1200. continue
  1201. }
  1202. instanceType, err := res.GetString("instance_type")
  1203. if err != nil {
  1204. log.Warnf("CostModel.ComputeAllocation: Node CPU cost query result missing field: \"%s\" for node \"%s\"", err, node)
  1205. continue
  1206. }
  1207. providerID, err := res.GetString("provider_id")
  1208. if err != nil {
  1209. log.Warnf("CostModel.ComputeAllocation: Node CPU cost query result missing field: \"%s\" for node \"%s\"", err, node)
  1210. continue
  1211. }
  1212. key := newNodeKey(cluster, node)
  1213. if _, ok := nodeMap[key]; !ok {
  1214. nodeMap[key] = &nodePricing{
  1215. Name: node,
  1216. NodeType: instanceType,
  1217. ProviderID: provider.ParseID(providerID),
  1218. }
  1219. }
  1220. nodeMap[key].CostPerCPUHr = res.Values[0].Value
  1221. }
  1222. }
  1223. func applyNodeCostPerRAMGiBHr(nodeMap map[nodeKey]*nodePricing, resNodeCostPerRAMGiBHr []*prom.QueryResult) {
  1224. for _, res := range resNodeCostPerRAMGiBHr {
  1225. cluster, err := res.GetString(env.GetPromClusterLabel())
  1226. if err != nil {
  1227. cluster = env.GetClusterID()
  1228. }
  1229. node, err := res.GetString("node")
  1230. if err != nil {
  1231. log.Warnf("CostModel.ComputeAllocation: Node RAM cost query result missing field: \"%s\" for node \"%s\"", err, node)
  1232. continue
  1233. }
  1234. instanceType, err := res.GetString("instance_type")
  1235. if err != nil {
  1236. log.Warnf("CostModel.ComputeAllocation: Node RAM cost query result missing field: \"%s\" for node \"%s\"", err, node)
  1237. continue
  1238. }
  1239. providerID, err := res.GetString("provider_id")
  1240. if err != nil {
  1241. log.Warnf("CostModel.ComputeAllocation: Node RAM cost query result missing field: \"%s\" for node \"%s\"", err, node)
  1242. continue
  1243. }
  1244. key := newNodeKey(cluster, node)
  1245. if _, ok := nodeMap[key]; !ok {
  1246. nodeMap[key] = &nodePricing{
  1247. Name: node,
  1248. NodeType: instanceType,
  1249. ProviderID: provider.ParseID(providerID),
  1250. }
  1251. }
  1252. nodeMap[key].CostPerRAMGiBHr = res.Values[0].Value
  1253. }
  1254. }
  1255. func applyNodeCostPerGPUHr(nodeMap map[nodeKey]*nodePricing, resNodeCostPerGPUHr []*prom.QueryResult) {
  1256. for _, res := range resNodeCostPerGPUHr {
  1257. cluster, err := res.GetString(env.GetPromClusterLabel())
  1258. if err != nil {
  1259. cluster = env.GetClusterID()
  1260. }
  1261. node, err := res.GetString("node")
  1262. if err != nil {
  1263. log.Warnf("CostModel.ComputeAllocation: Node GPU cost query result missing field: \"%s\" for node \"%s\"", err, node)
  1264. continue
  1265. }
  1266. instanceType, err := res.GetString("instance_type")
  1267. if err != nil {
  1268. log.Warnf("CostModel.ComputeAllocation: Node GPU cost query result missing field: \"%s\" for node \"%s\"", err, node)
  1269. continue
  1270. }
  1271. providerID, err := res.GetString("provider_id")
  1272. if err != nil {
  1273. log.Warnf("CostModel.ComputeAllocation: Node GPU cost query result missing field: \"%s\" for node \"%s\"", err, node)
  1274. continue
  1275. }
  1276. key := newNodeKey(cluster, node)
  1277. if _, ok := nodeMap[key]; !ok {
  1278. nodeMap[key] = &nodePricing{
  1279. Name: node,
  1280. NodeType: instanceType,
  1281. ProviderID: provider.ParseID(providerID),
  1282. }
  1283. }
  1284. nodeMap[key].CostPerGPUHr = res.Values[0].Value
  1285. }
  1286. }
  1287. func applyNodeSpot(nodeMap map[nodeKey]*nodePricing, resNodeIsSpot []*prom.QueryResult) {
  1288. for _, res := range resNodeIsSpot {
  1289. cluster, err := res.GetString(env.GetPromClusterLabel())
  1290. if err != nil {
  1291. cluster = env.GetClusterID()
  1292. }
  1293. node, err := res.GetString("node")
  1294. if err != nil {
  1295. log.Warnf("CostModel.ComputeAllocation: Node spot query result missing field: %s", err)
  1296. continue
  1297. }
  1298. key := newNodeKey(cluster, node)
  1299. if _, ok := nodeMap[key]; !ok {
  1300. log.Warnf("CostModel.ComputeAllocation: Node spot query result for missing node: %s", key)
  1301. continue
  1302. }
  1303. nodeMap[key].Preemptible = res.Values[0].Value > 0
  1304. }
  1305. }
  1306. func applyNodeDiscount(nodeMap map[nodeKey]*nodePricing, cm *CostModel) {
  1307. if cm == nil {
  1308. return
  1309. }
  1310. c, err := cm.Provider.GetConfig()
  1311. if err != nil {
  1312. log.Errorf("CostModel.ComputeAllocation: applyNodeDiscount: %s", err)
  1313. return
  1314. }
  1315. discount, err := ParsePercentString(c.Discount)
  1316. if err != nil {
  1317. log.Errorf("CostModel.ComputeAllocation: applyNodeDiscount: %s", err)
  1318. return
  1319. }
  1320. negotiatedDiscount, err := ParsePercentString(c.NegotiatedDiscount)
  1321. if err != nil {
  1322. log.Errorf("CostModel.ComputeAllocation: applyNodeDiscount: %s", err)
  1323. return
  1324. }
  1325. for _, node := range nodeMap {
  1326. // TODO GKE Reserved Instances into account
  1327. node.Discount = cm.Provider.CombinedDiscountForNode(node.NodeType, node.Preemptible, discount, negotiatedDiscount)
  1328. node.CostPerCPUHr *= (1.0 - node.Discount)
  1329. node.CostPerRAMGiBHr *= (1.0 - node.Discount)
  1330. }
  1331. }
  1332. func (cm *CostModel) applyNodesToPod(podMap map[podKey]*pod, nodeMap map[nodeKey]*nodePricing) {
  1333. for _, pod := range podMap {
  1334. for _, alloc := range pod.Allocations {
  1335. cluster := alloc.Properties.Cluster
  1336. nodeName := alloc.Properties.Node
  1337. thisNodeKey := newNodeKey(cluster, nodeName)
  1338. node := cm.getNodePricing(nodeMap, thisNodeKey)
  1339. alloc.Properties.ProviderID = node.ProviderID
  1340. alloc.CPUCost = alloc.CPUCoreHours * node.CostPerCPUHr
  1341. alloc.RAMCost = (alloc.RAMByteHours / 1024 / 1024 / 1024) * node.CostPerRAMGiBHr
  1342. alloc.GPUCost = alloc.GPUHours * node.CostPerGPUHr
  1343. }
  1344. }
  1345. }
  1346. // getCustomNodePricing converts the CostModel's configured custom pricing
  1347. // values into a nodePricing instance.
  1348. func (cm *CostModel) getCustomNodePricing(spot bool, providerID string) *nodePricing {
  1349. customPricingConfig, err := cm.Provider.GetConfig()
  1350. if err != nil {
  1351. return nil
  1352. }
  1353. cpuCostStr := customPricingConfig.CPU
  1354. gpuCostStr := customPricingConfig.GPU
  1355. ramCostStr := customPricingConfig.RAM
  1356. if spot {
  1357. cpuCostStr = customPricingConfig.SpotCPU
  1358. gpuCostStr = customPricingConfig.SpotGPU
  1359. ramCostStr = customPricingConfig.SpotRAM
  1360. }
  1361. node := &nodePricing{
  1362. Source: "custom",
  1363. ProviderID: providerID,
  1364. }
  1365. costPerCPUHr, err := strconv.ParseFloat(cpuCostStr, 64)
  1366. if err != nil {
  1367. log.Warnf("CostModel: custom pricing has illegal CPU cost: %s", cpuCostStr)
  1368. }
  1369. node.CostPerCPUHr = costPerCPUHr
  1370. costPerGPUHr, err := strconv.ParseFloat(gpuCostStr, 64)
  1371. if err != nil {
  1372. log.Warnf("CostModel: custom pricing has illegal GPU cost: %s", gpuCostStr)
  1373. }
  1374. node.CostPerGPUHr = costPerGPUHr
  1375. costPerRAMHr, err := strconv.ParseFloat(ramCostStr, 64)
  1376. if err != nil {
  1377. log.Warnf("CostModel: custom pricing has illegal RAM cost: %s", ramCostStr)
  1378. }
  1379. node.CostPerRAMGiBHr = costPerRAMHr
  1380. return node
  1381. }
  1382. // getNodePricing determines node pricing, given a key and a mapping from keys
  1383. // to their nodePricing instances, as well as the custom pricing configuration
  1384. // inherent to the CostModel instance. If custom pricing is set, use that. If
  1385. // not, use the pricing defined by the given key. If that doesn't exist, fall
  1386. // back on custom pricing as a default.
  1387. func (cm *CostModel) getNodePricing(nodeMap map[nodeKey]*nodePricing, nodeKey nodeKey) *nodePricing {
  1388. // Find the relevant nodePricing, if it exists. If not, substitute the
  1389. // custom nodePricing as a default.
  1390. node, ok := nodeMap[nodeKey]
  1391. if !ok || node == nil {
  1392. if nodeKey.Node != "" {
  1393. log.DedupedWarningf(5, "CostModel: failed to find node for %s", nodeKey)
  1394. }
  1395. // since the node pricing data is not found, and this won't change for the duration of the allocation
  1396. // build process, we can update the node map with the defaults to prevent future failed lookups
  1397. nodeMap[nodeKey] = cm.getCustomNodePricing(false, "")
  1398. return nodeMap[nodeKey]
  1399. }
  1400. // If custom pricing is enabled and can be retrieved, override detected
  1401. // node pricing with the custom values.
  1402. customPricingConfig, err := cm.Provider.GetConfig()
  1403. if err != nil {
  1404. log.Warnf("CostModel: failed to load custom pricing: %s", err)
  1405. }
  1406. if provider.CustomPricesEnabled(cm.Provider) && customPricingConfig != nil {
  1407. return cm.getCustomNodePricing(node.Preemptible, node.ProviderID)
  1408. }
  1409. node.Source = "prometheus"
  1410. // If any of the values are NaN or zero, replace them with the custom
  1411. // values as default.
  1412. // TODO:CLEANUP can't we parse these custom prices once? why do we store
  1413. // them as strings like this?
  1414. if node.CostPerCPUHr == 0 || math.IsNaN(node.CostPerCPUHr) {
  1415. log.Warnf("CostModel: node pricing has illegal CostPerCPUHr; replacing with custom pricing: %s", nodeKey)
  1416. cpuCostStr := customPricingConfig.CPU
  1417. if node.Preemptible {
  1418. cpuCostStr = customPricingConfig.SpotCPU
  1419. }
  1420. costPerCPUHr, err := strconv.ParseFloat(cpuCostStr, 64)
  1421. if err != nil {
  1422. log.Warnf("CostModel: custom pricing has illegal CPU cost: %s", cpuCostStr)
  1423. }
  1424. node.CostPerCPUHr = costPerCPUHr
  1425. node.Source += "/customCPU"
  1426. }
  1427. if math.IsNaN(node.CostPerGPUHr) {
  1428. log.Warnf("CostModel: node pricing has illegal CostPerGPUHr; replacing with custom pricing: %s", nodeKey)
  1429. gpuCostStr := customPricingConfig.GPU
  1430. if node.Preemptible {
  1431. gpuCostStr = customPricingConfig.SpotGPU
  1432. }
  1433. costPerGPUHr, err := strconv.ParseFloat(gpuCostStr, 64)
  1434. if err != nil {
  1435. log.Warnf("CostModel: custom pricing has illegal GPU cost: %s", gpuCostStr)
  1436. }
  1437. node.CostPerGPUHr = costPerGPUHr
  1438. node.Source += "/customGPU"
  1439. }
  1440. if node.CostPerRAMGiBHr == 0 || math.IsNaN(node.CostPerRAMGiBHr) {
  1441. log.Warnf("CostModel: node pricing has illegal CostPerRAMHr; replacing with custom pricing: %s", nodeKey)
  1442. ramCostStr := customPricingConfig.RAM
  1443. if node.Preemptible {
  1444. ramCostStr = customPricingConfig.SpotRAM
  1445. }
  1446. costPerRAMHr, err := strconv.ParseFloat(ramCostStr, 64)
  1447. if err != nil {
  1448. log.Warnf("CostModel: custom pricing has illegal RAM cost: %s", ramCostStr)
  1449. }
  1450. node.CostPerRAMGiBHr = costPerRAMHr
  1451. node.Source += "/customRAM"
  1452. }
  1453. return node
  1454. }
  1455. /* PV/PVC Helpers */
  1456. func buildPVMap(resolution time.Duration, pvMap map[pvKey]*pv, resPVCostPerGiBHour, resPVActiveMins []*prom.QueryResult) {
  1457. for _, result := range resPVActiveMins {
  1458. key, err := resultPVKey(result, env.GetPromClusterLabel(), "persistentvolume")
  1459. if err != nil {
  1460. log.Warnf("CostModel.ComputeAllocation: pv bytes query result missing field: %s", err)
  1461. continue
  1462. }
  1463. pvStart, pvEnd := calculateStartAndEnd(result, resolution)
  1464. if pvStart.IsZero() || pvEnd.IsZero() {
  1465. log.Warnf("CostModel.ComputeAllocation: pv %s has no running time", key)
  1466. }
  1467. pvMap[key] = &pv{
  1468. Cluster: key.Cluster,
  1469. Name: key.PersistentVolume,
  1470. Start: pvStart,
  1471. End: pvEnd,
  1472. }
  1473. }
  1474. for _, result := range resPVCostPerGiBHour {
  1475. key, err := resultPVKey(result, env.GetPromClusterLabel(), "volumename")
  1476. if err != nil {
  1477. log.Warnf("CostModel.ComputeAllocation: thisPV bytes query result missing field: %s", err)
  1478. continue
  1479. }
  1480. if _, ok := pvMap[key]; !ok {
  1481. pvMap[key] = &pv{
  1482. Cluster: key.Cluster,
  1483. Name: key.PersistentVolume,
  1484. }
  1485. }
  1486. pvMap[key].CostPerGiBHour = result.Values[0].Value
  1487. }
  1488. }
  1489. func applyPVBytes(pvMap map[pvKey]*pv, resPVBytes []*prom.QueryResult) {
  1490. for _, res := range resPVBytes {
  1491. key, err := resultPVKey(res, env.GetPromClusterLabel(), "persistentvolume")
  1492. if err != nil {
  1493. log.Warnf("CostModel.ComputeAllocation: pv bytes query result missing field: %s", err)
  1494. continue
  1495. }
  1496. if _, ok := pvMap[key]; !ok {
  1497. log.Warnf("CostModel.ComputeAllocation: pv bytes result for missing pv: %s", err)
  1498. continue
  1499. }
  1500. pvMap[key].Bytes = res.Values[0].Value
  1501. }
  1502. }
  1503. func buildPVCMap(resolution time.Duration, pvcMap map[pvcKey]*pvc, pvMap map[pvKey]*pv, resPVCInfo []*prom.QueryResult) {
  1504. for _, res := range resPVCInfo {
  1505. cluster, err := res.GetString(env.GetPromClusterLabel())
  1506. if err != nil {
  1507. cluster = env.GetClusterID()
  1508. }
  1509. values, err := res.GetStrings("persistentvolumeclaim", "storageclass", "volumename", "namespace")
  1510. if err != nil {
  1511. log.DedupedWarningf(10, "CostModel.ComputeAllocation: pvc info query result missing field: %s", err)
  1512. continue
  1513. }
  1514. namespace := values["namespace"]
  1515. name := values["persistentvolumeclaim"]
  1516. volume := values["volumename"]
  1517. storageClass := values["storageclass"]
  1518. pvKey := newPVKey(cluster, volume)
  1519. pvcKey := newPVCKey(cluster, namespace, name)
  1520. pvcStart, pvcEnd := calculateStartAndEnd(res, resolution)
  1521. if pvcStart.IsZero() || pvcEnd.IsZero() {
  1522. log.Warnf("CostModel.ComputeAllocation: pvc %s has no running time", pvcKey)
  1523. }
  1524. if _, ok := pvMap[pvKey]; !ok {
  1525. continue
  1526. }
  1527. pvMap[pvKey].StorageClass = storageClass
  1528. if _, ok := pvcMap[pvcKey]; !ok {
  1529. pvcMap[pvcKey] = &pvc{}
  1530. }
  1531. pvcMap[pvcKey].Name = name
  1532. pvcMap[pvcKey].Namespace = namespace
  1533. pvcMap[pvcKey].Cluster = cluster
  1534. pvcMap[pvcKey].Volume = pvMap[pvKey]
  1535. pvcMap[pvcKey].Start = pvcStart
  1536. pvcMap[pvcKey].End = pvcEnd
  1537. }
  1538. }
  1539. func applyPVCBytesRequested(pvcMap map[pvcKey]*pvc, resPVCBytesRequested []*prom.QueryResult) {
  1540. for _, res := range resPVCBytesRequested {
  1541. key, err := resultPVCKey(res, env.GetPromClusterLabel(), "namespace", "persistentvolumeclaim")
  1542. if err != nil {
  1543. continue
  1544. }
  1545. if _, ok := pvcMap[key]; !ok {
  1546. continue
  1547. }
  1548. pvcMap[key].Bytes = res.Values[0].Value
  1549. }
  1550. }
  1551. func buildPodPVCMap(podPVCMap map[podKey][]*pvc, pvMap map[pvKey]*pv, pvcMap map[pvcKey]*pvc, podMap map[podKey]*pod, resPodPVCAllocation []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey, ingestPodUID bool) {
  1552. for _, res := range resPodPVCAllocation {
  1553. cluster, err := res.GetString(env.GetPromClusterLabel())
  1554. if err != nil {
  1555. cluster = env.GetClusterID()
  1556. }
  1557. values, err := res.GetStrings("persistentvolume", "persistentvolumeclaim", "pod", "namespace")
  1558. if err != nil {
  1559. log.DedupedWarningf(5, "CostModel.ComputeAllocation: pvc allocation query result missing field: %s", err)
  1560. continue
  1561. }
  1562. namespace := values["namespace"]
  1563. pod := values["pod"]
  1564. name := values["persistentvolumeclaim"]
  1565. volume := values["persistentvolume"]
  1566. key := newPodKey(cluster, namespace, pod)
  1567. pvKey := newPVKey(cluster, volume)
  1568. pvcKey := newPVCKey(cluster, namespace, name)
  1569. var keys []podKey
  1570. if ingestPodUID {
  1571. if uidKeys, ok := podUIDKeyMap[key]; ok {
  1572. keys = append(keys, uidKeys...)
  1573. }
  1574. } else {
  1575. keys = []podKey{key}
  1576. }
  1577. for _, key := range keys {
  1578. if _, ok := pvMap[pvKey]; !ok {
  1579. log.DedupedWarningf(5, "CostModel.ComputeAllocation: pv missing for pvc allocation query result: %s", pvKey)
  1580. continue
  1581. }
  1582. if _, ok := podPVCMap[key]; !ok {
  1583. podPVCMap[key] = []*pvc{}
  1584. }
  1585. pvc, ok := pvcMap[pvcKey]
  1586. if !ok {
  1587. log.DedupedWarningf(5, "CostModel.ComputeAllocation: pvc missing for pvc allocation query: %s", pvcKey)
  1588. continue
  1589. }
  1590. if pod, ok := podMap[key]; !ok || len(pod.Allocations) <= 0 {
  1591. log.DedupedWarningf(10, "CostModel.ComputeAllocation: pvc %s for missing pod %s", pvcKey, key)
  1592. continue
  1593. }
  1594. pvc.Mounted = true
  1595. podPVCMap[key] = append(podPVCMap[key], pvc)
  1596. }
  1597. }
  1598. }
  1599. func applyPVCsToPods(window kubecost.Window, podMap map[podKey]*pod, podPVCMap map[podKey][]*pvc, pvcMap map[pvcKey]*pvc) {
  1600. // Because PVCs can be shared among pods, the respective pv cost
  1601. // needs to be evenly distributed to those pods based on time
  1602. // running, as well as the amount of time the pvc was shared.
  1603. // Build a relation between every pvc to the pods that mount it
  1604. // and a window representing the interval during which they
  1605. // were associated.
  1606. pvcPodWindowMap := make(map[pvcKey]map[podKey]kubecost.Window)
  1607. for thisPodKey, thisPod := range podMap {
  1608. if pvcs, ok := podPVCMap[thisPodKey]; ok {
  1609. for _, thisPVC := range pvcs {
  1610. // Determine the (start, end) of the relationship between the
  1611. // given pvc and the associated Allocation so that a precise
  1612. // number of hours can be used to compute cumulative cost.
  1613. s, e := thisPod.Start, thisPod.End
  1614. if thisPVC.Start.After(thisPod.Start) {
  1615. s = thisPVC.Start
  1616. }
  1617. if thisPVC.End.Before(thisPod.End) {
  1618. e = thisPVC.End
  1619. }
  1620. thisPVCKey := thisPVC.key()
  1621. if pvcPodWindowMap[thisPVCKey] == nil {
  1622. pvcPodWindowMap[thisPVCKey] = make(map[podKey]kubecost.Window)
  1623. }
  1624. pvcPodWindowMap[thisPVCKey][thisPodKey] = kubecost.NewWindow(&s, &e)
  1625. }
  1626. }
  1627. }
  1628. for thisPVCKey, podWindowMap := range pvcPodWindowMap {
  1629. // Build out a pv price coefficient for each pod with a pvc. Each
  1630. // pvc-pod relation needs a coefficient which modifies the pv cost
  1631. // such that pv costs can be shared between all pods using that pvc.
  1632. // Get single-point intervals from alloc-pvc relation windows.
  1633. intervals := getIntervalPointsFromWindows(podWindowMap)
  1634. pvc, ok := pvcMap[thisPVCKey]
  1635. if !ok {
  1636. log.DedupedWarningf(5, "Missing pvc with key %s", thisPVCKey)
  1637. continue
  1638. }
  1639. // Determine coefficients for each pvc-pod relation.
  1640. sharedPVCCostCoefficients := getPVCCostCoefficients(intervals, pvc)
  1641. // Distribute pvc costs to Allocations
  1642. for thisPodKey, coeffComponents := range sharedPVCCostCoefficients {
  1643. pod, ok2 := podMap[thisPodKey]
  1644. // If pod does not exist or the pod does not have any allocations
  1645. // get unmounted pod for cluster
  1646. if !ok2 || len(pod.Allocations) == 0 {
  1647. // Get namespace unmounted pod, as pvc will have a namespace
  1648. pod = getUnmountedPodForNamespace(window, podMap, pvc.Cluster, pvc.Namespace)
  1649. }
  1650. for _, alloc := range pod.Allocations {
  1651. s, e := pod.Start, pod.End
  1652. minutes := e.Sub(s).Minutes()
  1653. hrs := minutes / 60.0
  1654. gib := pvc.Bytes / 1024 / 1024 / 1024
  1655. cost := pvc.Volume.CostPerGiBHour * gib * hrs
  1656. byteHours := pvc.Bytes * hrs
  1657. coef := getCoefficientFromComponents(coeffComponents)
  1658. // Apply the size and cost of the pv to the allocation, each
  1659. // weighted by count (i.e. the number of containers in the pod)
  1660. // record the amount of total PVBytes Hours attributable to a given pv
  1661. if alloc.PVs == nil {
  1662. alloc.PVs = kubecost.PVAllocations{}
  1663. }
  1664. pvKey := kubecost.PVKey{
  1665. Cluster: pvc.Volume.Cluster,
  1666. Name: pvc.Volume.Name,
  1667. }
  1668. // Both Cost and byteHours should be multiplied by the coef and divided by count
  1669. // so that you if all allocations with a given pv key are summed the result of those
  1670. // would be equal to the values of the original pv
  1671. count := float64(len(pod.Allocations))
  1672. alloc.PVs[pvKey] = &kubecost.PVAllocation{
  1673. ByteHours: byteHours * coef / count,
  1674. Cost: cost * coef / count,
  1675. }
  1676. }
  1677. }
  1678. }
  1679. }
  1680. func applyUnmountedPVs(window kubecost.Window, podMap map[podKey]*pod, pvMap map[pvKey]*pv, pvcMap map[pvcKey]*pvc) {
  1681. for _, pv := range pvMap {
  1682. mounted := false
  1683. for _, pvc := range pvcMap {
  1684. if pvc.Volume == nil {
  1685. continue
  1686. }
  1687. if pvc.Volume == pv {
  1688. mounted = true
  1689. break
  1690. }
  1691. }
  1692. if !mounted {
  1693. // a pv without a pvc will not have a namespace, so get the cluster unmounted pod
  1694. pod := getUnmountedPodForCluster(window, podMap, pv.Cluster)
  1695. // Calculate pv Cost
  1696. // Unmounted pv should have correct keyso it can still reconcile
  1697. thisPVKey := kubecost.PVKey{
  1698. Cluster: pv.Cluster,
  1699. Name: pv.Name,
  1700. }
  1701. gib := pv.Bytes / 1024 / 1024 / 1024
  1702. hrs := pv.minutes() / 60.0
  1703. cost := pv.CostPerGiBHour * gib * hrs
  1704. unmountedPVs := kubecost.PVAllocations{
  1705. thisPVKey: {
  1706. ByteHours: pv.Bytes * hrs,
  1707. Cost: cost,
  1708. },
  1709. }
  1710. pod.Allocations[kubecost.UnmountedSuffix].PVs = pod.Allocations[kubecost.UnmountedSuffix].PVs.Add(unmountedPVs)
  1711. }
  1712. }
  1713. }
  1714. func applyUnmountedPVCs(window kubecost.Window, podMap map[podKey]*pod, pvcMap map[pvcKey]*pvc) {
  1715. for _, pvc := range pvcMap {
  1716. if !pvc.Mounted && pvc.Volume != nil {
  1717. // Get namespace unmounted pod, as pvc will have a namespace
  1718. pod := getUnmountedPodForNamespace(window, podMap, pvc.Cluster, pvc.Namespace)
  1719. // Calculate pv Cost
  1720. // Unmounted pv should have correct key so it can still reconcile
  1721. thisPVKey := kubecost.PVKey{
  1722. Cluster: pvc.Volume.Cluster,
  1723. Name: pvc.Volume.Name,
  1724. }
  1725. // Use the Volume Bytes here because pvc bytes could be different,
  1726. // however the pv bytes are what are going to determine cost
  1727. gib := pvc.Volume.Bytes / 1024 / 1024 / 1024
  1728. hrs := pvc.Volume.minutes() / 60.0
  1729. cost := pvc.Volume.CostPerGiBHour * gib * hrs
  1730. unmountedPVs := kubecost.PVAllocations{
  1731. thisPVKey: {
  1732. ByteHours: pvc.Volume.Bytes * hrs,
  1733. Cost: cost,
  1734. },
  1735. }
  1736. pod.Allocations[kubecost.UnmountedSuffix].PVs = pod.Allocations[kubecost.UnmountedSuffix].PVs.Add(unmountedPVs)
  1737. }
  1738. }
  1739. }
  1740. /* Helper Helpers */
  1741. // getUnmountedPodForCluster retrieve the unmounted pod for a cluster and create it if it does not exist
  1742. func getUnmountedPodForCluster(window kubecost.Window, podMap map[podKey]*pod, cluster string) *pod {
  1743. container := kubecost.UnmountedSuffix
  1744. podName := kubecost.UnmountedSuffix
  1745. namespace := kubecost.UnmountedSuffix
  1746. node := ""
  1747. thisPodKey := getUnmountedPodKey(cluster)
  1748. // Initialize pod and container if they do not already exist
  1749. thisPod, ok := podMap[thisPodKey]
  1750. if !ok {
  1751. thisPod = &pod{
  1752. Window: window.Clone(),
  1753. Start: *window.Start(),
  1754. End: *window.End(),
  1755. Key: thisPodKey,
  1756. Allocations: map[string]*kubecost.Allocation{},
  1757. }
  1758. thisPod.appendContainer(container)
  1759. thisPod.Allocations[container].Properties.Cluster = cluster
  1760. thisPod.Allocations[container].Properties.Node = node
  1761. thisPod.Allocations[container].Properties.Namespace = namespace
  1762. thisPod.Allocations[container].Properties.Pod = podName
  1763. thisPod.Allocations[container].Properties.Container = container
  1764. thisPod.Node = node
  1765. podMap[thisPodKey] = thisPod
  1766. }
  1767. return thisPod
  1768. }
  1769. // getUnmountedPodForNamespace is as getUnmountedPodForCluster, but keys allocation property pod/namespace field off namespace
  1770. // This creates or adds allocations to an unmounted pod in the specified namespace, rather than in __unmounted__
  1771. func getUnmountedPodForNamespace(window kubecost.Window, podMap map[podKey]*pod, cluster string, namespace string) *pod {
  1772. container := kubecost.UnmountedSuffix
  1773. podName := fmt.Sprintf("%s-unmounted-pvcs", namespace)
  1774. node := ""
  1775. thisPodKey := newPodKey(cluster, namespace, podName)
  1776. // Initialize pod and container if they do not already exist
  1777. thisPod, ok := podMap[thisPodKey]
  1778. if !ok {
  1779. thisPod = &pod{
  1780. Window: window.Clone(),
  1781. Start: *window.Start(),
  1782. End: *window.End(),
  1783. Key: thisPodKey,
  1784. Allocations: map[string]*kubecost.Allocation{},
  1785. }
  1786. thisPod.appendContainer(container)
  1787. thisPod.Allocations[container].Properties.Cluster = cluster
  1788. thisPod.Allocations[container].Properties.Node = node
  1789. thisPod.Allocations[container].Properties.Namespace = namespace
  1790. thisPod.Allocations[container].Properties.Pod = podName
  1791. thisPod.Allocations[container].Properties.Container = container
  1792. thisPod.Node = node
  1793. podMap[thisPodKey] = thisPod
  1794. }
  1795. return thisPod
  1796. }
  1797. func calculateStartAndEnd(result *prom.QueryResult, resolution time.Duration) (time.Time, time.Time) {
  1798. s := time.Unix(int64(result.Values[0].Timestamp), 0).UTC()
  1799. // subtract resolution from start time to cover full time period
  1800. s = s.Add(-resolution)
  1801. e := time.Unix(int64(result.Values[len(result.Values)-1].Timestamp), 0).UTC()
  1802. return s, e
  1803. }
  1804. // calculateStartEndFromIsRunning Calculates the start and end of a prom result when the values of the datum are 0 for not running and 1 for running
  1805. // the coeffs are used to adjust the start and end when the value is not equal to 1 or 0, which means that pod came up or went down in that window.
  1806. func calculateStartEndFromIsRunning(result *prom.QueryResult, resolution time.Duration, window kubecost.Window) (time.Time, time.Time) {
  1807. // start and end are the timestamps of the first and last
  1808. // minutes the pod was running, respectively. We subtract one resolution
  1809. // from start because this point will actually represent the end
  1810. // of the first minute. We don't subtract from end because it
  1811. // already represents the end of the last minute.
  1812. var start, end time.Time
  1813. startAdjustmentCoeff, endAdjustmentCoeff := 1.0, 1.0
  1814. for _, datum := range result.Values {
  1815. t := time.Unix(int64(datum.Timestamp), 0)
  1816. if start.IsZero() && datum.Value > 0 && window.Contains(t) {
  1817. // Set the start timestamp to the earliest non-zero timestamp
  1818. start = t
  1819. // Record adjustment coefficient, i.e. the portion of the start
  1820. // timestamp to "ignore". That is, sometimes the value will be
  1821. // 0.5, meaning that we should discount the time running by
  1822. // half of the resolution the timestamp stands for.
  1823. startAdjustmentCoeff = (1.0 - datum.Value)
  1824. }
  1825. if datum.Value > 0 && window.Contains(t) {
  1826. // Set the end timestamp to the latest non-zero timestamp
  1827. end = t
  1828. // Record adjustment coefficient, i.e. the portion of the end
  1829. // timestamp to "ignore". (See explanation above for start.)
  1830. endAdjustmentCoeff = (1.0 - datum.Value)
  1831. }
  1832. }
  1833. // Do not attempt to adjust start if it is zero
  1834. if !start.IsZero() {
  1835. // Adjust timestamps according to the resolution and the adjustment
  1836. // coefficients, as described above. That is, count the start timestamp
  1837. // from the beginning of the resolution, not the end. Then "reduce" the
  1838. // start and end by the correct amount, in the case that the "running"
  1839. // value of the first or last timestamp was not a full 1.0.
  1840. start = start.Add(-resolution)
  1841. // Note: the *100 and /100 are necessary because Duration is an int, so
  1842. // 0.5, for instance, will be truncated, resulting in no adjustment.
  1843. start = start.Add(time.Duration(startAdjustmentCoeff*100) * resolution / time.Duration(100))
  1844. end = end.Add(-time.Duration(endAdjustmentCoeff*100) * resolution / time.Duration(100))
  1845. // Ensure that the start is always within the window, adjusting
  1846. // for the occasions where start falls 1m before the query window.
  1847. // NOTE: window here will always be closed (so no need to nil check
  1848. // "start").
  1849. // TODO:CLEANUP revisit query methodology to figure out why this is
  1850. // happening on occasion
  1851. if start.Before(*window.Start()) {
  1852. start = *window.Start()
  1853. }
  1854. }
  1855. // do not attempt to adjust end if it is zero
  1856. if !end.IsZero() {
  1857. // If there is only one point with a value <= 0.5 that the start and
  1858. // end timestamps both share, then we will enter this case because at
  1859. // least half of a resolution will be subtracted from both the start
  1860. // and the end. If that is the case, then add back half of each side
  1861. // so that the pod is said to run for half a resolution total.
  1862. // e.g. For resolution 1m and a value of 0.5 at one timestamp, we'll
  1863. // end up with end == start and each coeff == 0.5. In
  1864. // that case, add 0.25m to each side, resulting in 0.5m duration.
  1865. if !end.After(start) {
  1866. start = start.Add(-time.Duration(50*startAdjustmentCoeff) * resolution / time.Duration(100))
  1867. end = end.Add(time.Duration(50*endAdjustmentCoeff) * resolution / time.Duration(100))
  1868. }
  1869. // Ensure that the allocEnf is always within the window, adjusting
  1870. // for the occasions where end falls 1m after the query window. This
  1871. // has not ever happened, but is symmetrical with the start check
  1872. // above.
  1873. // NOTE: window here will always be closed (so no need to nil check
  1874. // "end").
  1875. // TODO:CLEANUP revisit query methodology to figure out why this is
  1876. // happening on occasion
  1877. if end.After(*window.End()) {
  1878. end = *window.End()
  1879. }
  1880. }
  1881. return start, end
  1882. }