allocation_helpers.go 70 KB

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