allocation_helpers.go 72 KB

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