allocation.go 70 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129213021312132213321342135213621372138213921402141214221432144214521462147214821492150215121522153215421552156215721582159216021612162216321642165216621672168216921702171217221732174217521762177217821792180
  1. package kubecost
  2. import (
  3. "bytes"
  4. "fmt"
  5. "sort"
  6. "strings"
  7. "sync"
  8. "time"
  9. "github.com/kubecost/cost-model/pkg/log"
  10. "github.com/kubecost/cost-model/pkg/util"
  11. "github.com/kubecost/cost-model/pkg/util/json"
  12. )
  13. // TODO Clean-up use of IsEmpty; nil checks should be separated for safety.
  14. // TODO Consider making Allocation an interface, which is fulfilled by structs
  15. // like KubernetesAllocation, IdleAllocation, and ExternalAllocation.
  16. // ExternalSuffix indicates an external allocation
  17. const ExternalSuffix = "__external__"
  18. // IdleSuffix indicates an idle allocation property
  19. const IdleSuffix = "__idle__"
  20. // SharedSuffix indicates an shared allocation property
  21. const SharedSuffix = "__shared__"
  22. // UnallocatedSuffix indicates an unallocated allocation property
  23. const UnallocatedSuffix = "__unallocated__"
  24. // UnmountedSuffix indicated allocation to an unmounted PV
  25. const UnmountedSuffix = "__unmounted__"
  26. // ShareWeighted indicates that a shared resource should be shared as a
  27. // proportion of the cost of the remaining allocations.
  28. const ShareWeighted = "__weighted__"
  29. // ShareEven indicates that a shared resource should be shared evenly across
  30. // all remaining allocations.
  31. const ShareEven = "__even__"
  32. // ShareNone indicates that a shareable resource should not be shared
  33. const ShareNone = "__none__"
  34. // Allocation is a unit of resource allocation and cost for a given window
  35. // of time and for a given kubernetes construct with its associated set of
  36. // properties.
  37. // TODO:CLEANUP consider dropping name in favor of just AllocationProperties and an
  38. // Assets-style key() function for AllocationSet.
  39. type Allocation struct {
  40. Name string `json:"name"`
  41. Properties *AllocationProperties `json:"properties,omitempty"`
  42. Window Window `json:"window"`
  43. Start time.Time `json:"start"`
  44. End time.Time `json:"end"`
  45. CPUCoreHours float64 `json:"cpuCoreHours"`
  46. CPUCoreRequestAverage float64 `json:"cpuCoreRequestAverage"`
  47. CPUCoreUsageAverage float64 `json:"cpuCoreUsageAverage"`
  48. CPUCost float64 `json:"cpuCost"`
  49. CPUCostAdjustment float64 `json:"cpuCostAdjustment"`
  50. GPUHours float64 `json:"gpuHours"`
  51. GPUCost float64 `json:"gpuCost"`
  52. GPUCostAdjustment float64 `json:"gpuCostAdjustment"`
  53. NetworkCost float64 `json:"networkCost"`
  54. LoadBalancerCost float64 `json:"loadBalancerCost"`
  55. PVByteHours float64 `json:"pvByteHours"`
  56. PVCost float64 `json:"pvCost"`
  57. PVCostAdjustment float64 `json:"pvCostAdjustment"`
  58. RAMByteHours float64 `json:"ramByteHours"`
  59. RAMBytesRequestAverage float64 `json:"ramByteRequestAverage"`
  60. RAMBytesUsageAverage float64 `json:"ramByteUsageAverage"`
  61. RAMCost float64 `json:"ramCost"`
  62. RAMCostAdjustment float64 `json:"ramCostAdjustment"`
  63. SharedCost float64 `json:"sharedCost"`
  64. ExternalCost float64 `json:"externalCost"`
  65. // RawAllocationOnly is a pointer so if it is not present it will be
  66. // marshalled as null rather than as an object with Go default values.
  67. RawAllocationOnly *RawAllocationOnlyData `json:"rawAllocationOnly"`
  68. }
  69. // RawAllocationOnlyData is information that only belong in "raw" Allocations,
  70. // those which have not undergone aggregation, accumulation, or any other form
  71. // of combination to produce a new Allocation from other Allocations.
  72. //
  73. // Max usage data belongs here because computing the overall maximum from two
  74. // or more Allocations is a non-trivial operation that cannot be defined without
  75. // maintaining a large amount of state. Consider the following example:
  76. // _______________________________________________
  77. //
  78. // A1 Using 3 CPU ---- ----- ------
  79. // A2 Using 2 CPU ---- ----- ----
  80. // A3 Using 1 CPU --- --
  81. // _______________________________________________
  82. // Time ---->
  83. //
  84. // The logical maximum CPU usage is 5, but this cannot be calculated iteratively,
  85. // which is how we calculate aggregations and accumulations of Allocations currently.
  86. // This becomes a problem I could call "maximum sum of overlapping intervals" and is
  87. // essentially a variant of an interval scheduling algorithm.
  88. //
  89. // If we had types to differentiate between regular Allocations and AggregatedAllocations
  90. // then this type would be unnecessary and its fields would go into the regular Allocation
  91. // and not in the AggregatedAllocation.
  92. type RawAllocationOnlyData struct {
  93. CPUCoreUsageMax float64 `json:"cpuCoreUsageMax"`
  94. RAMBytesUsageMax float64 `json:"ramByteUsageMax"`
  95. }
  96. // AllocationMatchFunc is a function that can be used to match Allocations by
  97. // returning true for any given Allocation if a condition is met.
  98. type AllocationMatchFunc func(*Allocation) bool
  99. // Add returns the result of summing the two given Allocations, which sums the
  100. // summary fields (e.g. costs, resources) and recomputes efficiency. Neither of
  101. // the two original Allocations are mutated in the process.
  102. func (a *Allocation) Add(that *Allocation) (*Allocation, error) {
  103. if a == nil {
  104. return that.Clone(), nil
  105. }
  106. if that == nil {
  107. return a.Clone(), nil
  108. }
  109. // Note: no need to clone "that", as add only mutates the receiver
  110. agg := a.Clone()
  111. agg.add(that)
  112. return agg, nil
  113. }
  114. // Clone returns a deep copy of the given Allocation
  115. func (a *Allocation) Clone() *Allocation {
  116. if a == nil {
  117. return nil
  118. }
  119. return &Allocation{
  120. Name: a.Name,
  121. Properties: a.Properties.Clone(),
  122. Window: a.Window.Clone(),
  123. Start: a.Start,
  124. End: a.End,
  125. CPUCoreHours: a.CPUCoreHours,
  126. CPUCoreRequestAverage: a.CPUCoreRequestAverage,
  127. CPUCoreUsageAverage: a.CPUCoreUsageAverage,
  128. CPUCost: a.CPUCost,
  129. CPUCostAdjustment: a.CPUCostAdjustment,
  130. GPUHours: a.GPUHours,
  131. GPUCost: a.GPUCost,
  132. GPUCostAdjustment: a.GPUCostAdjustment,
  133. NetworkCost: a.NetworkCost,
  134. LoadBalancerCost: a.LoadBalancerCost,
  135. PVByteHours: a.PVByteHours,
  136. PVCost: a.PVCost,
  137. PVCostAdjustment: a.PVCostAdjustment,
  138. RAMByteHours: a.RAMByteHours,
  139. RAMBytesRequestAverage: a.RAMBytesRequestAverage,
  140. RAMBytesUsageAverage: a.RAMBytesUsageAverage,
  141. RAMCost: a.RAMCost,
  142. RAMCostAdjustment: a.RAMCostAdjustment,
  143. SharedCost: a.SharedCost,
  144. ExternalCost: a.ExternalCost,
  145. RawAllocationOnly: a.RawAllocationOnly.Clone(),
  146. }
  147. }
  148. // Clone returns a deep copy of the given RawAllocationOnlyData
  149. func (r *RawAllocationOnlyData) Clone() *RawAllocationOnlyData {
  150. if r == nil {
  151. return nil
  152. }
  153. return &RawAllocationOnlyData{
  154. CPUCoreUsageMax: r.CPUCoreUsageMax,
  155. RAMBytesUsageMax: r.RAMBytesUsageMax,
  156. }
  157. }
  158. // Equal returns true if the values held in the given Allocation precisely
  159. // match those of the receiving Allocation. nil does not match nil. Floating
  160. // point values need to match according to util.IsApproximately, which accounts
  161. // for small, reasonable floating point error margins.
  162. func (a *Allocation) Equal(that *Allocation) bool {
  163. if a == nil || that == nil {
  164. return false
  165. }
  166. if a.Name != that.Name {
  167. return false
  168. }
  169. if !a.Properties.Equal(that.Properties) {
  170. return false
  171. }
  172. if !a.Window.Equal(that.Window) {
  173. return false
  174. }
  175. if !a.Start.Equal(that.Start) {
  176. return false
  177. }
  178. if !a.End.Equal(that.End) {
  179. return false
  180. }
  181. if !util.IsApproximately(a.CPUCoreHours, that.CPUCoreHours) {
  182. return false
  183. }
  184. if !util.IsApproximately(a.CPUCost, that.CPUCost) {
  185. return false
  186. }
  187. if !util.IsApproximately(a.CPUCostAdjustment, that.CPUCostAdjustment) {
  188. return false
  189. }
  190. if !util.IsApproximately(a.GPUHours, that.GPUHours) {
  191. return false
  192. }
  193. if !util.IsApproximately(a.GPUCost, that.GPUCost) {
  194. return false
  195. }
  196. if !util.IsApproximately(a.GPUCostAdjustment, that.GPUCostAdjustment) {
  197. return false
  198. }
  199. if !util.IsApproximately(a.NetworkCost, that.NetworkCost) {
  200. return false
  201. }
  202. if !util.IsApproximately(a.LoadBalancerCost, that.LoadBalancerCost) {
  203. return false
  204. }
  205. if !util.IsApproximately(a.PVByteHours, that.PVByteHours) {
  206. return false
  207. }
  208. if !util.IsApproximately(a.PVCost, that.PVCost) {
  209. return false
  210. }
  211. if !util.IsApproximately(a.PVCostAdjustment, that.PVCostAdjustment) {
  212. return false
  213. }
  214. if !util.IsApproximately(a.RAMByteHours, that.RAMByteHours) {
  215. return false
  216. }
  217. if !util.IsApproximately(a.RAMCost, that.RAMCost) {
  218. return false
  219. }
  220. if !util.IsApproximately(a.RAMCostAdjustment, that.RAMCostAdjustment) {
  221. return false
  222. }
  223. if !util.IsApproximately(a.SharedCost, that.SharedCost) {
  224. return false
  225. }
  226. if !util.IsApproximately(a.ExternalCost, that.ExternalCost) {
  227. return false
  228. }
  229. if a.RawAllocationOnly == nil && that.RawAllocationOnly != nil {
  230. return false
  231. }
  232. if a.RawAllocationOnly != nil && that.RawAllocationOnly == nil {
  233. return false
  234. }
  235. if a.RawAllocationOnly != nil && that.RawAllocationOnly != nil {
  236. if !util.IsApproximately(a.RawAllocationOnly.CPUCoreUsageMax, that.RawAllocationOnly.CPUCoreUsageMax) {
  237. return false
  238. }
  239. if !util.IsApproximately(a.RawAllocationOnly.RAMBytesUsageMax, that.RawAllocationOnly.RAMBytesUsageMax) {
  240. return false
  241. }
  242. }
  243. return true
  244. }
  245. // TotalCost is the total cost of the Allocation
  246. func (a *Allocation) TotalCost() float64 {
  247. return a.CPUTotalCost() + a.GPUTotalCost() + a.RAMTotalCost() + a.PVTotalCost() + a.NetworkCost + a.SharedCost + a.ExternalCost + a.LoadBalancerCost
  248. }
  249. func (a *Allocation) CPUTotalCost() float64 {
  250. return a.CPUCost + a.CPUCostAdjustment
  251. }
  252. func (a *Allocation) GPUTotalCost() float64 {
  253. return a.GPUCost + a.GPUCostAdjustment
  254. }
  255. func (a *Allocation) RAMTotalCost() float64 {
  256. return a.RAMCost + a.RAMCostAdjustment
  257. }
  258. func (a *Allocation) PVTotalCost() float64 {
  259. return a.PVCost + a.PVCostAdjustment
  260. }
  261. // CPUEfficiency is the ratio of usage to request. If there is no request and
  262. // no usage or cost, then efficiency is zero. If there is no request, but there
  263. // is usage or cost, then efficiency is 100%.
  264. func (a *Allocation) CPUEfficiency() float64 {
  265. if a.CPUCoreRequestAverage > 0 {
  266. return a.CPUCoreUsageAverage / a.CPUCoreRequestAverage
  267. }
  268. if a.CPUCoreUsageAverage == 0.0 || a.CPUCost == 0.0 {
  269. return 0.0
  270. }
  271. return 1.0
  272. }
  273. // RAMEfficiency is the ratio of usage to request. If there is no request and
  274. // no usage or cost, then efficiency is zero. If there is no request, but there
  275. // is usage or cost, then efficiency is 100%.
  276. func (a *Allocation) RAMEfficiency() float64 {
  277. if a.RAMBytesRequestAverage > 0 {
  278. return a.RAMBytesUsageAverage / a.RAMBytesRequestAverage
  279. }
  280. if a.RAMBytesUsageAverage == 0.0 || a.RAMCost == 0.0 {
  281. return 0.0
  282. }
  283. return 1.0
  284. }
  285. // TotalEfficiency is the cost-weighted average of CPU and RAM efficiency. If
  286. // there is no cost at all, then efficiency is zero.
  287. func (a *Allocation) TotalEfficiency() float64 {
  288. if a.RAMTotalCost()+a.CPUTotalCost() > 0 {
  289. ramCostEff := a.RAMEfficiency() * a.RAMTotalCost()
  290. cpuCostEff := a.CPUEfficiency() * a.CPUTotalCost()
  291. return (ramCostEff + cpuCostEff) / (a.CPUTotalCost() + a.RAMTotalCost())
  292. }
  293. return 0.0
  294. }
  295. // CPUCores converts the Allocation's CPUCoreHours into average CPUCores
  296. func (a *Allocation) CPUCores() float64 {
  297. if a.Minutes() <= 0.0 {
  298. return 0.0
  299. }
  300. return a.CPUCoreHours / (a.Minutes() / 60.0)
  301. }
  302. // RAMBytes converts the Allocation's RAMByteHours into average RAMBytes
  303. func (a *Allocation) RAMBytes() float64 {
  304. if a.Minutes() <= 0.0 {
  305. return 0.0
  306. }
  307. return a.RAMByteHours / (a.Minutes() / 60.0)
  308. }
  309. // GPUs converts the Allocation's GPUHours into average GPUs
  310. func (a *Allocation) GPUs() float64 {
  311. if a.Minutes() <= 0.0 {
  312. return 0.0
  313. }
  314. return a.GPUHours / (a.Minutes() / 60.0)
  315. }
  316. // PVBytes converts the Allocation's PVByteHours into average PVBytes
  317. func (a *Allocation) PVBytes() float64 {
  318. if a.Minutes() <= 0.0 {
  319. return 0.0
  320. }
  321. return a.PVByteHours / (a.Minutes() / 60.0)
  322. }
  323. // MarshalJSON implements json.Marshaler interface
  324. func (a *Allocation) MarshalJSON() ([]byte, error) {
  325. buffer := bytes.NewBufferString("{")
  326. jsonEncodeString(buffer, "name", a.Name, ",")
  327. jsonEncode(buffer, "properties", a.Properties, ",")
  328. jsonEncode(buffer, "window", a.Window, ",")
  329. jsonEncodeString(buffer, "start", a.Start.Format(time.RFC3339), ",")
  330. jsonEncodeString(buffer, "end", a.End.Format(time.RFC3339), ",")
  331. jsonEncodeFloat64(buffer, "minutes", a.Minutes(), ",")
  332. jsonEncodeFloat64(buffer, "cpuCores", a.CPUCores(), ",")
  333. jsonEncodeFloat64(buffer, "cpuCoreRequestAverage", a.CPUCoreRequestAverage, ",")
  334. jsonEncodeFloat64(buffer, "cpuCoreUsageAverage", a.CPUCoreUsageAverage, ",")
  335. jsonEncodeFloat64(buffer, "cpuCoreHours", a.CPUCoreHours, ",")
  336. jsonEncodeFloat64(buffer, "cpuCost", a.CPUCost, ",")
  337. jsonEncodeFloat64(buffer, "cpuCostAdjustment", a.CPUCostAdjustment, ",")
  338. jsonEncodeFloat64(buffer, "cpuEfficiency", a.CPUEfficiency(), ",")
  339. jsonEncodeFloat64(buffer, "gpuCount", a.GPUs(), ",")
  340. jsonEncodeFloat64(buffer, "gpuHours", a.GPUHours, ",")
  341. jsonEncodeFloat64(buffer, "gpuCost", a.GPUCost, ",")
  342. jsonEncodeFloat64(buffer, "gpuCostAdjustment", a.GPUCostAdjustment, ",")
  343. jsonEncodeFloat64(buffer, "networkCost", a.NetworkCost, ",")
  344. jsonEncodeFloat64(buffer, "loadBalancerCost", a.LoadBalancerCost, ",")
  345. jsonEncodeFloat64(buffer, "pvBytes", a.PVBytes(), ",")
  346. jsonEncodeFloat64(buffer, "pvByteHours", a.PVByteHours, ",")
  347. jsonEncodeFloat64(buffer, "pvCost", a.PVCost, ",")
  348. jsonEncodeFloat64(buffer, "pvCostAdjustment", a.PVCostAdjustment, ",")
  349. jsonEncodeFloat64(buffer, "ramBytes", a.RAMBytes(), ",")
  350. jsonEncodeFloat64(buffer, "ramByteRequestAverage", a.RAMBytesRequestAverage, ",")
  351. jsonEncodeFloat64(buffer, "ramByteUsageAverage", a.RAMBytesUsageAverage, ",")
  352. jsonEncodeFloat64(buffer, "ramByteHours", a.RAMByteHours, ",")
  353. jsonEncodeFloat64(buffer, "ramCost", a.RAMCost, ",")
  354. jsonEncodeFloat64(buffer, "ramCostAdjustment", a.RAMCostAdjustment, ",")
  355. jsonEncodeFloat64(buffer, "ramEfficiency", a.RAMEfficiency(), ",")
  356. jsonEncodeFloat64(buffer, "sharedCost", a.SharedCost, ",")
  357. jsonEncodeFloat64(buffer, "externalCost", a.ExternalCost, ",")
  358. jsonEncodeFloat64(buffer, "totalCost", a.TotalCost(), ",")
  359. jsonEncodeFloat64(buffer, "totalEfficiency", a.TotalEfficiency(), ",")
  360. jsonEncode(buffer, "rawAllocationOnly", a.RawAllocationOnly, "")
  361. buffer.WriteString("}")
  362. return buffer.Bytes(), nil
  363. }
  364. // Resolution returns the duration of time covered by the Allocation
  365. func (a *Allocation) Resolution() time.Duration {
  366. return a.End.Sub(a.Start)
  367. }
  368. // IsAggregated is true if the given Allocation has been aggregated, which we
  369. // define by a lack of AllocationProperties.
  370. func (a *Allocation) IsAggregated() bool {
  371. return a == nil || a.Properties == nil
  372. }
  373. // IsExternal is true if the given Allocation represents external costs.
  374. func (a *Allocation) IsExternal() bool {
  375. return strings.Contains(a.Name, ExternalSuffix)
  376. }
  377. // IsIdle is true if the given Allocation represents idle costs.
  378. func (a *Allocation) IsIdle() bool {
  379. return strings.Contains(a.Name, IdleSuffix)
  380. }
  381. // IsUnallocated is true if the given Allocation represents unallocated costs.
  382. func (a *Allocation) IsUnallocated() bool {
  383. return strings.Contains(a.Name, UnallocatedSuffix)
  384. }
  385. // Minutes returns the number of minutes the Allocation represents, as defined
  386. // by the difference between the end and start times.
  387. func (a *Allocation) Minutes() float64 {
  388. return a.End.Sub(a.Start).Minutes()
  389. }
  390. // Share adds the TotalCost of the given Allocation to the SharedCost of the
  391. // receiving Allocation. No Start, End, Window, or AllocationProperties are considered.
  392. // Neither Allocation is mutated; a new Allocation is always returned.
  393. func (a *Allocation) Share(that *Allocation) (*Allocation, error) {
  394. if that == nil {
  395. return a.Clone(), nil
  396. }
  397. if a == nil {
  398. return nil, fmt.Errorf("cannot share with nil Allocation")
  399. }
  400. agg := a.Clone()
  401. agg.SharedCost += that.TotalCost()
  402. return agg, nil
  403. }
  404. // String represents the given Allocation as a string
  405. func (a *Allocation) String() string {
  406. return fmt.Sprintf("%s%s=%.2f", a.Name, NewWindow(&a.Start, &a.End), a.TotalCost())
  407. }
  408. func (a *Allocation) add(that *Allocation) {
  409. if a == nil {
  410. log.Warningf("Allocation.AggregateBy: trying to add a nil receiver")
  411. return
  412. }
  413. // Preserve string properties that are matching between the two allocations
  414. a.Properties = a.Properties.Intersection(that.Properties)
  415. // Expand the window to encompass both Allocations
  416. a.Window = a.Window.Expand(that.Window)
  417. // Sum non-cumulative fields by turning them into cumulative, adding them,
  418. // and then converting them back into averages after minutes have been
  419. // combined (just below).
  420. cpuReqCoreMins := a.CPUCoreRequestAverage * a.Minutes()
  421. cpuReqCoreMins += that.CPUCoreRequestAverage * that.Minutes()
  422. cpuUseCoreMins := a.CPUCoreUsageAverage * a.Minutes()
  423. cpuUseCoreMins += that.CPUCoreUsageAverage * that.Minutes()
  424. ramReqByteMins := a.RAMBytesRequestAverage * a.Minutes()
  425. ramReqByteMins += that.RAMBytesRequestAverage * that.Minutes()
  426. ramUseByteMins := a.RAMBytesUsageAverage * a.Minutes()
  427. ramUseByteMins += that.RAMBytesUsageAverage * that.Minutes()
  428. // Expand Start and End to be the "max" of among the given Allocations
  429. if that.Start.Before(a.Start) {
  430. a.Start = that.Start
  431. }
  432. if that.End.After(a.End) {
  433. a.End = that.End
  434. }
  435. // Convert cumulative request and usage back into rates
  436. // TODO:TEST write a unit test that fails if this is done incorrectly
  437. if a.Minutes() > 0 {
  438. a.CPUCoreRequestAverage = cpuReqCoreMins / a.Minutes()
  439. a.CPUCoreUsageAverage = cpuUseCoreMins / a.Minutes()
  440. a.RAMBytesRequestAverage = ramReqByteMins / a.Minutes()
  441. a.RAMBytesUsageAverage = ramUseByteMins / a.Minutes()
  442. } else {
  443. a.CPUCoreRequestAverage = 0.0
  444. a.CPUCoreUsageAverage = 0.0
  445. a.RAMBytesRequestAverage = 0.0
  446. a.RAMBytesUsageAverage = 0.0
  447. }
  448. // Sum all cumulative resource fields
  449. a.CPUCoreHours += that.CPUCoreHours
  450. a.GPUHours += that.GPUHours
  451. a.RAMByteHours += that.RAMByteHours
  452. a.PVByteHours += that.PVByteHours
  453. // Sum all cumulative cost fields
  454. a.CPUCost += that.CPUCost
  455. a.GPUCost += that.GPUCost
  456. a.RAMCost += that.RAMCost
  457. a.PVCost += that.PVCost
  458. a.NetworkCost += that.NetworkCost
  459. a.LoadBalancerCost += that.LoadBalancerCost
  460. a.SharedCost += that.SharedCost
  461. a.ExternalCost += that.ExternalCost
  462. // Sum all cumulative adjustment fields
  463. a.CPUCostAdjustment += that.CPUCostAdjustment
  464. a.RAMCostAdjustment += that.RAMCostAdjustment
  465. a.GPUCostAdjustment += that.GPUCostAdjustment
  466. a.PVCostAdjustment += that.PVCostAdjustment
  467. // Any data that is in a "raw allocation only" is not valid in any
  468. // sort of cumulative Allocation (like one that is added).
  469. a.RawAllocationOnly = nil
  470. }
  471. // AllocationSet stores a set of Allocations, each with a unique name, that share
  472. // a window. An AllocationSet is mutable, so treat it like a threadsafe map.
  473. type AllocationSet struct {
  474. sync.RWMutex
  475. allocations map[string]*Allocation
  476. externalKeys map[string]bool
  477. idleKeys map[string]bool
  478. Window Window
  479. Warnings []string
  480. Errors []string
  481. }
  482. // NewAllocationSet instantiates a new AllocationSet and, optionally, inserts
  483. // the given list of Allocations
  484. func NewAllocationSet(start, end time.Time, allocs ...*Allocation) *AllocationSet {
  485. as := &AllocationSet{
  486. allocations: map[string]*Allocation{},
  487. externalKeys: map[string]bool{},
  488. idleKeys: map[string]bool{},
  489. Window: NewWindow(&start, &end),
  490. }
  491. for _, a := range allocs {
  492. as.Insert(a)
  493. }
  494. return as
  495. }
  496. // AllocationAggregationOptions provide advanced functionality to AggregateBy, including
  497. // filtering results and sharing allocations. FilterFuncs are a list of match
  498. // functions such that, if any function fails, the allocation is ignored.
  499. // ShareFuncs are a list of match functions such that, if any function
  500. // succeeds, the allocation is marked as a shared resource. ShareIdle is a
  501. // simple flag for sharing idle resources.
  502. type AllocationAggregationOptions struct {
  503. FilterFuncs []AllocationMatchFunc
  504. SplitIdle bool
  505. MergeUnallocated bool
  506. ShareFuncs []AllocationMatchFunc
  507. ShareIdle string
  508. ShareSplit string
  509. SharedHourlyCosts map[string]float64
  510. }
  511. // AggregateBy aggregates the Allocations in the given AllocationSet by the given
  512. // AllocationProperty. This will only be legal if the AllocationSet is divisible by the
  513. // given AllocationProperty; e.g. Containers can be divided by Namespace, but not vice-a-versa.
  514. func (as *AllocationSet) AggregateBy(aggregateBy []string, options *AllocationAggregationOptions) error {
  515. // The order of operations for aggregating allocations is as follows:
  516. // 1. Partition external, idle, and shared allocations into separate sets.
  517. // Also, create the aggSet into which the results will be aggregated.
  518. // 2. Compute sharing coefficients for idle and shared resources
  519. // a) if idle allocation is to be shared, compute idle coefficients
  520. // b) if idle allocation is NOT shared, but filters are present, compute
  521. // idle filtration coefficients for the purpose of only returning the
  522. // portion of idle allocation that would have been shared with the
  523. // unfiltered results. (See unit tests 5.a,b,c)
  524. // c) generate shared allocation for then given shared overhead, which
  525. // must happen after (2a) and (2b)
  526. // d) if there are shared resources, compute share coefficients
  527. // 3. Drop any allocation that fails any of the filters
  528. // 4. Distribute idle allocations according to the idle coefficients
  529. // 5. Generate aggregation key and insert allocation into the output set
  530. // 6. If idle is shared and resources are shared, some idle might be shared
  531. // with a shared resource. Distribute that to the shared resources
  532. // prior to sharing them with the aggregated results.
  533. // 7. Apply idle filtration coefficients from step (2b)
  534. // 8. Distribute shared allocations according to the share coefficients.
  535. // 9. If there are external allocations that can be aggregated into
  536. // the output (i.e. they can be used to generate a valid key for
  537. // the given properties) then aggregate; otherwise... ignore them?
  538. // 10. If the merge idle option is enabled, merge any remaining idle
  539. // allocations into a single idle allocation
  540. if options == nil {
  541. options = &AllocationAggregationOptions{}
  542. }
  543. if as.IsEmpty() {
  544. return nil
  545. }
  546. // aggSet will collect the aggregated allocations
  547. aggSet := &AllocationSet{
  548. Window: as.Window.Clone(),
  549. }
  550. // externalSet will collect external allocations
  551. externalSet := &AllocationSet{
  552. Window: as.Window.Clone(),
  553. }
  554. // idleSet will be shared among aggSet after initial aggregation
  555. // is complete
  556. idleSet := &AllocationSet{
  557. Window: as.Window.Clone(),
  558. }
  559. // shareSet will be shared among aggSet after initial aggregation
  560. // is complete
  561. shareSet := &AllocationSet{
  562. Window: as.Window.Clone(),
  563. }
  564. as.Lock()
  565. defer as.Unlock()
  566. // (1) Loop and find all of the external, idle, and shared allocations. Add
  567. // them to their respective sets, removing them from the set of allocations
  568. // to aggregate.
  569. for _, alloc := range as.allocations {
  570. // External allocations get aggregated post-hoc (see step 6) and do
  571. // not necessarily contain complete sets of properties, so they are
  572. // moved to a separate AllocationSet.
  573. if alloc.IsExternal() {
  574. delete(as.externalKeys, alloc.Name)
  575. delete(as.allocations, alloc.Name)
  576. externalSet.Insert(alloc)
  577. continue
  578. }
  579. // Idle allocations should be separated into idleSet if they are to be
  580. // shared later on. If they are not to be shared, then add them to the
  581. // aggSet like any other allocation.
  582. if alloc.IsIdle() {
  583. delete(as.idleKeys, alloc.Name)
  584. delete(as.allocations, alloc.Name)
  585. if options.ShareIdle == ShareEven || options.ShareIdle == ShareWeighted {
  586. idleSet.Insert(alloc)
  587. } else {
  588. aggSet.Insert(alloc)
  589. }
  590. continue
  591. }
  592. // Shared allocations must be identified and separated prior to
  593. // aggregation and filtering. That is, if any of the ShareFuncs return
  594. // true for the allocation, then move it to shareSet.
  595. for _, sf := range options.ShareFuncs {
  596. if sf(alloc) {
  597. delete(as.idleKeys, alloc.Name)
  598. delete(as.allocations, alloc.Name)
  599. shareSet.Insert(alloc)
  600. break
  601. }
  602. }
  603. }
  604. // It's possible that no more un-shared, non-idle, non-external allocations
  605. // remain at this point. This always results in an emptySet, so return early.
  606. if len(as.allocations) == 0 {
  607. emptySet := &AllocationSet{
  608. Window: as.Window.Clone(),
  609. }
  610. as.allocations = emptySet.allocations
  611. return nil
  612. }
  613. // (2) In order to correctly share idle and shared costs, we first compute
  614. // sharing coefficients, which represent the proportion of each cost to
  615. // share with each allocation. Idle allocations are shared per-cluster,
  616. // per-allocation, and per-resource, while shared resources are shared per-
  617. // allocation only.
  618. //
  619. // For an idleCoefficient example, the entries:
  620. // [cluster1][cluster1/namespace1/pod1/container1][cpu] = 0.166667
  621. // [cluster1][cluster1/namespace1/pod1/container1][gpu] = 0.166667
  622. // [cluster1][cluster1/namespace1/pod1/container1][ram] = 0.687500
  623. // mean that the allocation "cluster1/namespace1/pod1/container1" will
  624. // receive 16.67% of cluster1's idle CPU and GPU costs and 68.75% of its
  625. // RAM costs.
  626. //
  627. // For a shareCoefficient example, the entries:
  628. // [namespace2] = 0.666667
  629. // [__filtered__] = 0.333333
  630. // mean that the post-aggregation allocation "namespace2" will receive
  631. // 66.67% of the shared resource costs, while the remaining 33.33% will
  632. // be filtered out, as they were shared with allocations that did not pass
  633. // one of the given filters.
  634. //
  635. // In order to maintain stable results when multiple operations are being
  636. // carried out (e.g. sharing idle, sharing resources, and filtering) these
  637. // coefficients are computed for the full set of allocations prior to
  638. // adding shared overhead and prior to applying filters.
  639. var err error
  640. // (2a) If there are idle costs to be shared, compute the coefficients for
  641. // sharing them among the non-idle, non-aggregated allocations (including
  642. // the shared allocations).
  643. var idleCoefficients map[string]map[string]map[string]float64
  644. if idleSet.Length() > 0 && options.ShareIdle != ShareNone {
  645. idleCoefficients, err = computeIdleCoeffs(options, as, shareSet)
  646. if err != nil {
  647. log.Warningf("AllocationSet.AggregateBy: compute idle coeff: %s", err)
  648. return fmt.Errorf("error computing idle coefficients: %s", err)
  649. }
  650. }
  651. // (2b) If idle costs are not to be shared, but there are filters, then we
  652. // need to track the amount of each idle allocation to "filter" in order to
  653. // maintain parity with the results when idle is shared. That is, we want
  654. // to return only the idle costs that would have been shared with the given
  655. // results, even if the filter had not been applied.
  656. //
  657. // For example, consider these results from aggregating by namespace with
  658. // two clusters:
  659. //
  660. // namespace1: 25.00
  661. // namespace2: 30.00
  662. // namespace3: 15.00
  663. // idle: 30.00
  664. //
  665. // When we then filter by cluster==cluster1, namespaces 2 and 3 are
  666. // reduced by the amount that existed on cluster2. Then, idle must also be
  667. // reduced by the relevant amount:
  668. //
  669. // namespace1: 25.00
  670. // namespace2: 15.00
  671. // idle: 20.00
  672. //
  673. // Note that this can happen for any field, not just cluster, so we again
  674. // need to track this on a per-cluster, per-allocation, per-resource basis.
  675. var idleFiltrationCoefficients map[string]map[string]map[string]float64
  676. if len(options.FilterFuncs) > 0 && options.ShareIdle == ShareNone {
  677. idleFiltrationCoefficients, err = computeIdleCoeffs(options, as, shareSet)
  678. if err != nil {
  679. return fmt.Errorf("error computing idle filtration coefficients: %s", err)
  680. }
  681. }
  682. // (2c) Convert SharedHourlyCosts to Allocations in the shareSet. This must
  683. // come after idle coefficients are computes so that allocations generated
  684. // by shared overhead do not skew the idle coefficient computation.
  685. for name, cost := range options.SharedHourlyCosts {
  686. if cost > 0.0 {
  687. hours := as.Resolution().Hours()
  688. // If set ends in the future, adjust hours accordingly
  689. diff := time.Now().Sub(as.End())
  690. if diff < 0.0 {
  691. hours += diff.Hours()
  692. }
  693. totalSharedCost := cost * hours
  694. shareSet.Insert(&Allocation{
  695. Name: fmt.Sprintf("%s/%s", name, SharedSuffix),
  696. Start: as.Start(),
  697. End: as.End(),
  698. SharedCost: totalSharedCost,
  699. Properties: &AllocationProperties{Cluster: SharedSuffix}, // The allocation needs to belong to a cluster,but it really doesn't matter which one, so just make it clear.
  700. })
  701. }
  702. }
  703. // (2d) Compute share coefficients for shared resources. These are computed
  704. // after idle coefficients, and are computed for the aggregated allocations
  705. // of the main allocation set. See above for details and an example.
  706. var shareCoefficients map[string]float64
  707. if shareSet.Length() > 0 {
  708. shareCoefficients, err = computeShareCoeffs(aggregateBy, options, as)
  709. if err != nil {
  710. return fmt.Errorf("error computing share coefficients: %s", err)
  711. }
  712. }
  713. // (3-5) Filter, distribute idle cost, and aggregate (in that order)
  714. for _, alloc := range as.allocations {
  715. cluster := alloc.Properties.Cluster
  716. if cluster == "" {
  717. log.Warningf("AllocationSet.AggregateBy: missing cluster for allocation: %s", alloc.Name)
  718. return fmt.Errorf("ClusterProp is not set")
  719. }
  720. skip := false
  721. // (3) If any of the filter funcs fail, immediately skip the allocation.
  722. for _, ff := range options.FilterFuncs {
  723. if !ff(alloc) {
  724. skip = true
  725. break
  726. }
  727. }
  728. if skip {
  729. // If we are tracking idle filtration coefficients, delete the
  730. // entry corresponding to the filtered allocation. (Deleting the
  731. // entry will result in that proportional amount being removed
  732. // from the idle allocation at the end of the process.)
  733. if idleFiltrationCoefficients != nil {
  734. if ifcc, ok := idleFiltrationCoefficients[cluster]; ok {
  735. delete(ifcc, alloc.Name)
  736. }
  737. }
  738. continue
  739. }
  740. // (4) Distribute idle allocations according to the idle coefficients
  741. // NOTE: if idle allocation is off (i.e. ShareIdle == ShareNone) then
  742. // all idle allocations will be in the aggSet at this point, so idleSet
  743. // will be empty and we won't enter this block.
  744. if idleSet.Length() > 0 {
  745. // Distribute idle allocations by coefficient per-cluster, per-allocation
  746. for _, idleAlloc := range idleSet.allocations {
  747. // Only share idle if the cluster matches; i.e. the allocation
  748. // is from the same cluster as the idle costs
  749. idleCluster := idleAlloc.Properties.Cluster
  750. if idleCluster == "" {
  751. return fmt.Errorf("ClusterProp is not set")
  752. }
  753. if idleCluster != cluster {
  754. continue
  755. }
  756. // Make sure idle coefficients exist
  757. if _, ok := idleCoefficients[cluster]; !ok {
  758. log.Warningf("AllocationSet.AggregateBy: error getting idle coefficient: no cluster '%s' for '%s'", cluster, alloc.Name)
  759. continue
  760. }
  761. if _, ok := idleCoefficients[cluster][alloc.Name]; !ok {
  762. log.Warningf("AllocationSet.AggregateBy: error getting idle coefficient for '%s'", alloc.Name)
  763. continue
  764. }
  765. alloc.CPUCoreHours += idleAlloc.CPUCoreHours * idleCoefficients[cluster][alloc.Name]["cpu"]
  766. alloc.GPUHours += idleAlloc.GPUHours * idleCoefficients[cluster][alloc.Name]["gpu"]
  767. alloc.RAMByteHours += idleAlloc.RAMByteHours * idleCoefficients[cluster][alloc.Name]["ram"]
  768. idleCPUCost := idleAlloc.CPUCost * idleCoefficients[cluster][alloc.Name]["cpu"]
  769. idleGPUCost := idleAlloc.GPUCost * idleCoefficients[cluster][alloc.Name]["gpu"]
  770. idleRAMCost := idleAlloc.RAMCost * idleCoefficients[cluster][alloc.Name]["ram"]
  771. alloc.CPUCost += idleCPUCost
  772. alloc.GPUCost += idleGPUCost
  773. alloc.RAMCost += idleRAMCost
  774. }
  775. }
  776. // (5) generate key to use for aggregation-by-key and allocation name
  777. key := alloc.generateKey(aggregateBy)
  778. alloc.Name = key
  779. if options.MergeUnallocated && alloc.IsUnallocated() {
  780. alloc.Name = UnallocatedSuffix
  781. }
  782. // Inserting the allocation with the generated key for a name will
  783. // perform the actual basic aggregation step.
  784. aggSet.Insert(alloc)
  785. }
  786. // (6) If idle is shared and resources are shared, it's possible that some
  787. // amount of idle cost will be shared with a shared resource. Distribute
  788. // that idle allocation, if it exists, to the respective shared allocations
  789. // before sharing with the aggregated allocations.
  790. if idleSet.Length() > 0 && shareSet.Length() > 0 {
  791. for _, alloc := range shareSet.allocations {
  792. cluster := alloc.Properties.Cluster
  793. if cluster == "" {
  794. log.Warningf("AllocationSet.AggregateBy: missing cluster for allocation: %s", alloc.Name)
  795. return err
  796. }
  797. // Distribute idle allocations by coefficient per-cluster, per-allocation
  798. for _, idleAlloc := range idleSet.allocations {
  799. // Only share idle if the cluster matches; i.e. the allocation
  800. // is from the same cluster as the idle costs
  801. idleCluster := idleAlloc.Properties.Cluster
  802. if idleCluster == "" {
  803. return fmt.Errorf("ClusterProp is not set")
  804. }
  805. if idleCluster != cluster {
  806. continue
  807. }
  808. // Make sure idle coefficients exist
  809. if _, ok := idleCoefficients[cluster]; !ok {
  810. log.Warningf("AllocationSet.AggregateBy: error getting idle coefficient: no cluster '%s' for '%s'", cluster, alloc.Name)
  811. continue
  812. }
  813. if _, ok := idleCoefficients[cluster][alloc.Name]; !ok {
  814. log.Warningf("AllocationSet.AggregateBy: error getting idle coefficient for '%s'", alloc.Name)
  815. continue
  816. }
  817. alloc.CPUCoreHours += idleAlloc.CPUCoreHours * idleCoefficients[cluster][alloc.Name]["cpu"]
  818. alloc.GPUHours += idleAlloc.GPUHours * idleCoefficients[cluster][alloc.Name]["gpu"]
  819. alloc.RAMByteHours += idleAlloc.RAMByteHours * idleCoefficients[cluster][alloc.Name]["ram"]
  820. idleCPUCost := idleAlloc.CPUCost * idleCoefficients[cluster][alloc.Name]["cpu"]
  821. idleGPUCost := idleAlloc.GPUCost * idleCoefficients[cluster][alloc.Name]["gpu"]
  822. idleRAMCost := idleAlloc.RAMCost * idleCoefficients[cluster][alloc.Name]["ram"]
  823. alloc.CPUCost += idleCPUCost
  824. alloc.GPUCost += idleGPUCost
  825. alloc.RAMCost += idleRAMCost
  826. }
  827. }
  828. }
  829. // clusterIdleFiltrationCoeffs is used to track per-resource idle
  830. // coefficients on a cluster-by-cluster basis. It is, essentailly, an
  831. // aggregation of idleFiltrationCoefficients after they have been
  832. // filtered above (in step 3)
  833. var clusterIdleFiltrationCoeffs map[string]map[string]float64
  834. if idleFiltrationCoefficients != nil {
  835. clusterIdleFiltrationCoeffs = map[string]map[string]float64{}
  836. for cluster, m := range idleFiltrationCoefficients {
  837. if _, ok := clusterIdleFiltrationCoeffs[cluster]; !ok {
  838. clusterIdleFiltrationCoeffs[cluster] = map[string]float64{
  839. "cpu": 0.0,
  840. "gpu": 0.0,
  841. "ram": 0.0,
  842. }
  843. }
  844. for _, n := range m {
  845. for resource, val := range n {
  846. clusterIdleFiltrationCoeffs[cluster][resource] += val
  847. }
  848. }
  849. }
  850. }
  851. // (7) If we have both un-shared idle allocations and idle filtration
  852. // coefficients then apply those. See step (2b) for an example.
  853. if len(aggSet.idleKeys) > 0 && clusterIdleFiltrationCoeffs != nil {
  854. for idleKey := range aggSet.idleKeys {
  855. idleAlloc := aggSet.Get(idleKey)
  856. cluster := idleAlloc.Properties.Cluster
  857. if cluster == "" {
  858. log.Warningf("AllocationSet.AggregateBy: idle allocation without cluster: %s", idleAlloc)
  859. continue
  860. }
  861. if resourceCoeffs, ok := clusterIdleFiltrationCoeffs[cluster]; ok {
  862. idleAlloc.CPUCost *= resourceCoeffs["cpu"]
  863. idleAlloc.CPUCoreHours *= resourceCoeffs["cpu"]
  864. idleAlloc.RAMCost *= resourceCoeffs["ram"]
  865. idleAlloc.RAMByteHours *= resourceCoeffs["ram"]
  866. }
  867. }
  868. }
  869. // (8) Distribute shared allocations according to the share coefficients.
  870. if shareSet.Length() > 0 {
  871. for _, alloc := range aggSet.allocations {
  872. for _, sharedAlloc := range shareSet.allocations {
  873. if _, ok := shareCoefficients[alloc.Name]; !ok {
  874. log.Warningf("AllocationSet.AggregateBy: error getting share coefficienct for '%s'", alloc.Name)
  875. continue
  876. }
  877. alloc.SharedCost += sharedAlloc.TotalCost() * shareCoefficients[alloc.Name]
  878. }
  879. }
  880. }
  881. // (9) Aggregate external allocations into aggregated allocations. This may
  882. // not be possible for every external allocation, but attempt to find an
  883. // exact key match, given each external allocation's proerties, and
  884. // aggregate if an exact match is found.
  885. for _, alloc := range externalSet.allocations {
  886. skip := false
  887. for _, ff := range options.FilterFuncs {
  888. if !ff(alloc) {
  889. skip = true
  890. break
  891. }
  892. }
  893. if !skip {
  894. key := alloc.generateKey(aggregateBy)
  895. alloc.Name = key
  896. aggSet.Insert(alloc)
  897. }
  898. }
  899. // (10) Combine all idle allocations into a single "__idle__" allocation
  900. if !options.SplitIdle {
  901. for _, idleAlloc := range aggSet.IdleAllocations() {
  902. aggSet.Delete(idleAlloc.Name)
  903. idleAlloc.Name = IdleSuffix
  904. aggSet.Insert(idleAlloc)
  905. }
  906. }
  907. as.allocations = aggSet.allocations
  908. return nil
  909. }
  910. func computeShareCoeffs(aggregateBy []string, options *AllocationAggregationOptions, as *AllocationSet) (map[string]float64, error) {
  911. // Compute coeffs by totalling per-allocation, then dividing by the total.
  912. coeffs := map[string]float64{}
  913. // Compute totals for all allocations
  914. total := 0.0
  915. // ShareEven counts each aggregation with even weight, whereas ShareWeighted
  916. // counts each aggregation proportionally to its respective costs
  917. shareType := options.ShareSplit
  918. // Record allocation values first, then normalize by totals to get percentages
  919. for _, alloc := range as.allocations {
  920. if alloc.IsIdle() {
  921. // Skip idle allocations in coefficient calculation
  922. continue
  923. }
  924. // Determine the post-aggregation key under which the allocation will
  925. // be shared.
  926. name := alloc.generateKey(aggregateBy)
  927. // If the current allocation will be filtered out in step 3, contribute
  928. // its share of the shared coefficient to a "__filtered__" bin, which
  929. // will ultimately be dropped. This step ensures that the shared cost
  930. // of a non-filtered allocation will be conserved even when the filter
  931. // is removed. (Otherwise, all the shared cost will get redistributed
  932. // over the unfiltered results, inflating their shared costs.)
  933. filtered := false
  934. for _, ff := range options.FilterFuncs {
  935. if !ff(alloc) {
  936. filtered = true
  937. break
  938. }
  939. }
  940. if filtered {
  941. name = "__filtered__"
  942. }
  943. if shareType == ShareEven {
  944. // Even distribution is not additive - set to 1.0 for everything
  945. coeffs[name] = 1.0
  946. // Total for even distribution is always the number of coefficients
  947. total = float64(len(coeffs))
  948. } else {
  949. // Both are additive for weighted distribution, where each
  950. // cumulative coefficient will be divided by the total.
  951. coeffs[name] += alloc.TotalCost()
  952. total += alloc.TotalCost()
  953. }
  954. }
  955. // Normalize coefficients by totals
  956. for a := range coeffs {
  957. if coeffs[a] > 0 && total > 0 {
  958. coeffs[a] /= total
  959. } else {
  960. log.Warningf("ETL: invalid values for shared coefficients: %d, %d", coeffs[a], total)
  961. coeffs[a] = 0.0
  962. }
  963. }
  964. return coeffs, nil
  965. }
  966. func computeIdleCoeffs(options *AllocationAggregationOptions, as *AllocationSet, shareSet *AllocationSet) (map[string]map[string]map[string]float64, error) {
  967. types := []string{"cpu", "gpu", "ram"}
  968. // Compute idle coefficients, then save them in AllocationAggregationOptions
  969. coeffs := map[string]map[string]map[string]float64{}
  970. // Compute totals per resource for CPU, GPU, RAM, and PV
  971. totals := map[string]map[string]float64{}
  972. // ShareEven counts each allocation with even weight, whereas ShareWeighted
  973. // counts each allocation proportionally to its respective costs
  974. shareType := options.ShareIdle
  975. // Record allocation values first, then normalize by totals to get percentages
  976. for _, alloc := range as.allocations {
  977. if alloc.IsIdle() {
  978. // Skip idle allocations in coefficient calculation
  979. continue
  980. }
  981. // We need to key the allocations by cluster id
  982. clusterID := alloc.Properties.Cluster
  983. if clusterID == "" {
  984. return nil, fmt.Errorf("ClusterProp is not set")
  985. }
  986. // get the name key for the allocation
  987. name := alloc.Name
  988. // Create cluster based tables if they don't exist
  989. if _, ok := coeffs[clusterID]; !ok {
  990. coeffs[clusterID] = map[string]map[string]float64{}
  991. }
  992. if _, ok := totals[clusterID]; !ok {
  993. totals[clusterID] = map[string]float64{}
  994. }
  995. if _, ok := coeffs[clusterID][name]; !ok {
  996. coeffs[clusterID][name] = map[string]float64{}
  997. }
  998. if shareType == ShareEven {
  999. for _, r := range types {
  1000. // Not additive - hard set to 1.0
  1001. coeffs[clusterID][name][r] = 1.0
  1002. // totals are additive
  1003. totals[clusterID][r] += 1.0
  1004. }
  1005. } else {
  1006. coeffs[clusterID][name]["cpu"] += alloc.CPUTotalCost()
  1007. coeffs[clusterID][name]["gpu"] += alloc.GPUTotalCost()
  1008. coeffs[clusterID][name]["ram"] += alloc.RAMTotalCost()
  1009. totals[clusterID]["cpu"] += alloc.CPUTotalCost()
  1010. totals[clusterID]["gpu"] += alloc.GPUTotalCost()
  1011. totals[clusterID]["ram"] += alloc.RAMTotalCost()
  1012. }
  1013. }
  1014. // Do the same for shared allocations
  1015. for _, alloc := range shareSet.allocations {
  1016. if alloc.IsIdle() {
  1017. // Skip idle allocations in coefficient calculation
  1018. continue
  1019. }
  1020. // We need to key the allocations by cluster id
  1021. clusterID := alloc.Properties.Cluster
  1022. if clusterID == "" {
  1023. return nil, fmt.Errorf("ClusterProp is not set")
  1024. }
  1025. // get the name key for the allocation
  1026. name := alloc.Name
  1027. // Create cluster based tables if they don't exist
  1028. if _, ok := coeffs[clusterID]; !ok {
  1029. coeffs[clusterID] = map[string]map[string]float64{}
  1030. }
  1031. if _, ok := totals[clusterID]; !ok {
  1032. totals[clusterID] = map[string]float64{}
  1033. }
  1034. if _, ok := coeffs[clusterID][name]; !ok {
  1035. coeffs[clusterID][name] = map[string]float64{}
  1036. }
  1037. if shareType == ShareEven {
  1038. for _, r := range types {
  1039. // Not additive - hard set to 1.0
  1040. coeffs[clusterID][name][r] = 1.0
  1041. // totals are additive
  1042. totals[clusterID][r] += 1.0
  1043. }
  1044. } else {
  1045. coeffs[clusterID][name]["cpu"] += alloc.CPUTotalCost()
  1046. coeffs[clusterID][name]["gpu"] += alloc.GPUTotalCost()
  1047. coeffs[clusterID][name]["ram"] += alloc.RAMTotalCost()
  1048. totals[clusterID]["cpu"] += alloc.CPUTotalCost()
  1049. totals[clusterID]["gpu"] += alloc.GPUTotalCost()
  1050. totals[clusterID]["ram"] += alloc.RAMTotalCost()
  1051. }
  1052. }
  1053. // Normalize coefficients by totals
  1054. for c := range coeffs {
  1055. for a := range coeffs[c] {
  1056. for _, r := range types {
  1057. if coeffs[c][a][r] > 0 && totals[c][r] > 0 {
  1058. coeffs[c][a][r] /= totals[c][r]
  1059. }
  1060. }
  1061. }
  1062. }
  1063. return coeffs, nil
  1064. }
  1065. func (a *Allocation) generateKey(aggregateBy []string) string {
  1066. if a == nil {
  1067. return ""
  1068. }
  1069. // Names will ultimately be joined into a single name, which uniquely
  1070. // identifies allocations.
  1071. names := []string{}
  1072. for _, agg := range aggregateBy {
  1073. switch true {
  1074. case agg == AllocationClusterProp:
  1075. names = append(names, a.Properties.Cluster)
  1076. case agg == AllocationNodeProp:
  1077. names = append(names, a.Properties.Node)
  1078. case agg == AllocationNamespaceProp:
  1079. names = append(names, a.Properties.Namespace)
  1080. case agg == AllocationControllerKindProp:
  1081. controllerKind := a.Properties.ControllerKind
  1082. if controllerKind == "" {
  1083. // Indicate that allocation has no controller
  1084. controllerKind = UnallocatedSuffix
  1085. }
  1086. names = append(names, controllerKind)
  1087. case agg == AllocationDaemonSetProp || agg == AllocationStatefulSetProp || agg == AllocationDeploymentProp || agg == AllocationJobProp:
  1088. controller := a.Properties.Controller
  1089. if agg != a.Properties.ControllerKind || controller == "" {
  1090. // The allocation does not have the specified controller kind
  1091. controller = UnallocatedSuffix
  1092. }
  1093. names = append(names, controller)
  1094. case agg == AllocationControllerProp:
  1095. controller := a.Properties.Controller
  1096. if controller == "" {
  1097. // Indicate that allocation has no controller
  1098. controller = UnallocatedSuffix
  1099. } else if a.Properties.ControllerKind != "" {
  1100. controller = fmt.Sprintf("%s:%s", a.Properties.ControllerKind, controller)
  1101. }
  1102. names = append(names, controller)
  1103. case agg == AllocationPodProp:
  1104. names = append(names, a.Properties.Pod)
  1105. case agg == AllocationContainerProp:
  1106. names = append(names, a.Properties.Container)
  1107. case agg == AllocationServiceProp:
  1108. services := a.Properties.Services
  1109. if services == nil || len(services) == 0 {
  1110. // Indicate that allocation has no services
  1111. names = append(names, UnallocatedSuffix)
  1112. } else {
  1113. // This just uses the first service
  1114. for _, service := range services {
  1115. names = append(names, service)
  1116. break
  1117. }
  1118. }
  1119. case strings.HasPrefix(agg, "label:"):
  1120. labels := a.Properties.Labels
  1121. if labels == nil {
  1122. // Indicate that allocation has no labels
  1123. names = append(names, UnallocatedSuffix)
  1124. } else {
  1125. labelNames := []string{}
  1126. aggLabels := strings.Split(strings.TrimPrefix(agg, "label:"), ";")
  1127. for _, labelName := range aggLabels {
  1128. if val, ok := labels[labelName]; ok {
  1129. labelNames = append(labelNames, fmt.Sprintf("%s=%s", labelName, val))
  1130. } else if indexOf(UnallocatedSuffix, labelNames) == -1 { // if UnallocatedSuffix not already in names
  1131. labelNames = append(labelNames, UnallocatedSuffix)
  1132. }
  1133. }
  1134. // resolve arbitrary ordering. e.g., app=app0/env=env0 is the same agg as env=env0/app=app0
  1135. if len(labelNames) > 1 {
  1136. sort.Strings(labelNames)
  1137. }
  1138. unallocatedSuffixIndex := indexOf(UnallocatedSuffix, labelNames)
  1139. // suffix should be at index 0 if it exists b/c of underscores
  1140. if unallocatedSuffixIndex != -1 {
  1141. labelNames = append(labelNames[:unallocatedSuffixIndex], labelNames[unallocatedSuffixIndex+1:]...)
  1142. labelNames = append(labelNames, UnallocatedSuffix) // append to end
  1143. }
  1144. names = append(names, labelNames...)
  1145. }
  1146. case strings.HasPrefix(agg, "annotation:"):
  1147. annotations := a.Properties.Annotations
  1148. if annotations == nil {
  1149. // Indicate that allocation has no annotations
  1150. names = append(names, UnallocatedSuffix)
  1151. } else {
  1152. annotationNames := []string{}
  1153. aggAnnotations := strings.Split(strings.TrimPrefix(agg, "annotation:"), ";")
  1154. for _, annotationName := range aggAnnotations {
  1155. if val, ok := annotations[annotationName]; ok {
  1156. annotationNames = append(annotationNames, fmt.Sprintf("%s=%s", annotationName, val))
  1157. } else if indexOf(UnallocatedSuffix, annotationNames) == -1 { // if UnallocatedSuffix not already in names
  1158. annotationNames = append(annotationNames, UnallocatedSuffix)
  1159. }
  1160. }
  1161. // resolve arbitrary ordering. e.g., app=app0/env=env0 is the same agg as env=env0/app=app0
  1162. if len(annotationNames) > 1 {
  1163. sort.Strings(annotationNames)
  1164. }
  1165. unallocatedSuffixIndex := indexOf(UnallocatedSuffix, annotationNames)
  1166. // suffix should be at index 0 if it exists b/c of underscores
  1167. if unallocatedSuffixIndex != -1 {
  1168. annotationNames = append(annotationNames[:unallocatedSuffixIndex], annotationNames[unallocatedSuffixIndex+1:]...)
  1169. annotationNames = append(annotationNames, UnallocatedSuffix) // append to end
  1170. }
  1171. names = append(names, annotationNames...)
  1172. }
  1173. }
  1174. }
  1175. return strings.Join(names, "/")
  1176. }
  1177. // TODO:CLEANUP get rid of this
  1178. // Helper function to check for slice membership. Not sure if repeated elsewhere in our codebase.
  1179. func indexOf(v string, arr []string) int {
  1180. for i, s := range arr {
  1181. // This is caseless equivalence
  1182. if strings.EqualFold(v, s) {
  1183. return i
  1184. }
  1185. }
  1186. return -1
  1187. }
  1188. // Clone returns a new AllocationSet with a deep copy of the given
  1189. // AllocationSet's allocations.
  1190. func (as *AllocationSet) Clone() *AllocationSet {
  1191. if as == nil {
  1192. return nil
  1193. }
  1194. as.RLock()
  1195. defer as.RUnlock()
  1196. allocs := map[string]*Allocation{}
  1197. for k, v := range as.allocations {
  1198. allocs[k] = v.Clone()
  1199. }
  1200. externalKeys := map[string]bool{}
  1201. for k, v := range as.externalKeys {
  1202. externalKeys[k] = v
  1203. }
  1204. idleKeys := map[string]bool{}
  1205. for k, v := range as.idleKeys {
  1206. idleKeys[k] = v
  1207. }
  1208. return &AllocationSet{
  1209. allocations: allocs,
  1210. externalKeys: externalKeys,
  1211. idleKeys: idleKeys,
  1212. Window: as.Window.Clone(),
  1213. }
  1214. }
  1215. // ComputeIdleAllocations computes the idle allocations for the AllocationSet,
  1216. // given a set of Assets. Ideally, assetSet should contain only Nodes, but if
  1217. // it contains other Assets, they will be ignored; only CPU, GPU and RAM are
  1218. // considered for idle allocation. If the Nodes have adjustments, then apply
  1219. // the adjustments proportionally to each of the resources so that total
  1220. // allocation with idle reflects the adjusted node costs. One idle allocation
  1221. // per-cluster will be computed and returned, keyed by cluster_id.
  1222. func (as *AllocationSet) ComputeIdleAllocations(assetSet *AssetSet) (map[string]*Allocation, error) {
  1223. if as == nil {
  1224. return nil, fmt.Errorf("cannot compute idle allocation for nil AllocationSet")
  1225. }
  1226. if assetSet == nil {
  1227. return nil, fmt.Errorf("cannot compute idle allocation with nil AssetSet")
  1228. }
  1229. if !as.Window.Equal(assetSet.Window) {
  1230. return nil, fmt.Errorf("cannot compute idle allocation for sets with mismatched windows: %s != %s", as.Window, assetSet.Window)
  1231. }
  1232. window := as.Window
  1233. // Build a map of cumulative cluster asset costs, per resource; i.e.
  1234. // cluster-to-{cpu|gpu|ram}-to-cost.
  1235. assetClusterResourceCosts := map[string]map[string]float64{}
  1236. assetSet.Each(func(key string, a Asset) {
  1237. if node, ok := a.(*Node); ok {
  1238. if _, ok := assetClusterResourceCosts[node.Properties().Cluster]; !ok {
  1239. assetClusterResourceCosts[node.Properties().Cluster] = map[string]float64{}
  1240. }
  1241. // adjustmentRate is used to scale resource costs proportionally
  1242. // by the adjustment. This is necessary because we only get one
  1243. // adjustment per Node, not one per-resource-per-Node.
  1244. //
  1245. // e.g. total cost = $90, adjustment = -$10 => 0.9
  1246. // e.g. total cost = $150, adjustment = -$300 => 0.3333
  1247. // e.g. total cost = $150, adjustment = $50 => 1.5
  1248. adjustmentRate := 1.0
  1249. if node.TotalCost()-node.Adjustment() == 0 {
  1250. // If (totalCost - adjustment) is 0.0 then adjustment cancels
  1251. // the entire node cost and we should make everything 0
  1252. // without dividing by 0.
  1253. adjustmentRate = 0.0
  1254. } else if node.Adjustment() != 0.0 {
  1255. // adjustmentRate is the ratio of cost-with-adjustment (i.e. TotalCost)
  1256. // to cost-without-adjustment (i.e. TotalCost - Adjustment).
  1257. adjustmentRate = node.TotalCost() / (node.TotalCost() - node.Adjustment())
  1258. }
  1259. cpuCost := node.CPUCost * (1.0 - node.Discount) * adjustmentRate
  1260. gpuCost := node.GPUCost * (1.0 - node.Discount) * adjustmentRate
  1261. ramCost := node.RAMCost * (1.0 - node.Discount) * adjustmentRate
  1262. assetClusterResourceCosts[node.Properties().Cluster]["cpu"] += cpuCost
  1263. assetClusterResourceCosts[node.Properties().Cluster]["gpu"] += gpuCost
  1264. assetClusterResourceCosts[node.Properties().Cluster]["ram"] += ramCost
  1265. }
  1266. })
  1267. // Determine start, end on a per-cluster basis
  1268. clusterStarts := map[string]time.Time{}
  1269. clusterEnds := map[string]time.Time{}
  1270. // Subtract allocated costs from asset costs, leaving only the remaining
  1271. // idle costs.
  1272. as.Each(func(name string, a *Allocation) {
  1273. cluster := a.Properties.Cluster
  1274. if cluster == "" {
  1275. // Failed to find allocation's cluster
  1276. return
  1277. }
  1278. if _, ok := assetClusterResourceCosts[cluster]; !ok {
  1279. // Failed to find assets for allocation's cluster
  1280. return
  1281. }
  1282. // Set cluster (start, end) if they are either not currently set,
  1283. // or if the detected (start, end) of the current allocation falls
  1284. // before or after, respectively, the current values.
  1285. if s, ok := clusterStarts[cluster]; !ok || a.Start.Before(s) {
  1286. clusterStarts[cluster] = a.Start
  1287. }
  1288. if e, ok := clusterEnds[cluster]; !ok || a.End.After(e) {
  1289. clusterEnds[cluster] = a.End
  1290. }
  1291. assetClusterResourceCosts[cluster]["cpu"] -= a.CPUTotalCost()
  1292. assetClusterResourceCosts[cluster]["gpu"] -= a.GPUTotalCost()
  1293. assetClusterResourceCosts[cluster]["ram"] -= a.RAMTotalCost()
  1294. })
  1295. // Turn remaining un-allocated asset costs into idle allocations
  1296. idleAllocs := map[string]*Allocation{}
  1297. for cluster, resources := range assetClusterResourceCosts {
  1298. // Default start and end to the (start, end) of the given window, but
  1299. // use the actual, detected (start, end) pair if they are available.
  1300. start := *window.Start()
  1301. if s, ok := clusterStarts[cluster]; ok && window.Contains(s) {
  1302. start = s
  1303. }
  1304. end := *window.End()
  1305. if e, ok := clusterEnds[cluster]; ok && window.Contains(e) {
  1306. end = e
  1307. }
  1308. idleAlloc := &Allocation{
  1309. Name: fmt.Sprintf("%s/%s", cluster, IdleSuffix),
  1310. Window: window.Clone(),
  1311. Properties: &AllocationProperties{Cluster: cluster},
  1312. Start: start,
  1313. End: end,
  1314. CPUCost: resources["cpu"],
  1315. GPUCost: resources["gpu"],
  1316. RAMCost: resources["ram"],
  1317. }
  1318. // Do not continue if multiple idle allocations are computed for a
  1319. // single cluster.
  1320. if _, ok := idleAllocs[cluster]; ok {
  1321. return nil, fmt.Errorf("duplicate idle allocations for cluster %s", cluster)
  1322. }
  1323. idleAllocs[cluster] = idleAlloc
  1324. }
  1325. return idleAllocs, nil
  1326. }
  1327. // Reconcile calculate the exact cost of Allocation by resource(cpu, ram, gpu etc) based on Asset(s) on which
  1328. // the Allocation depends.
  1329. func (as *AllocationSet) Reconcile(assetSet *AssetSet) error {
  1330. if as == nil {
  1331. return fmt.Errorf("cannot reconcile allocation for nil AllocationSet")
  1332. }
  1333. if assetSet == nil {
  1334. return fmt.Errorf("cannot reconcile allocation with nil AssetSet")
  1335. }
  1336. if !as.Window.Equal(assetSet.Window) {
  1337. return fmt.Errorf("cannot reconcile allocation for sets with mismatched windows: %s != %s", as.Window, assetSet.Window)
  1338. }
  1339. // Build map of Assets with type Node by their ProviderId so that they can be matched to Allocations to determine
  1340. // proper CPU GPU and RAM prices
  1341. nodeByProviderID := map[string]*Node{}
  1342. diskByName := map[string]*Disk{}
  1343. assetSet.Each(func(key string, a Asset) {
  1344. if node, ok := a.(*Node); ok && node.properties.ProviderID != "" {
  1345. nodeByProviderID[node.properties.ProviderID] = node
  1346. }
  1347. if disk, ok := a.(*Disk); ok {
  1348. diskByName[disk.properties.Name] = disk
  1349. }
  1350. })
  1351. // Match Assets against allocations and adjust allocation cost based on the proportion of the asset that they used
  1352. as.Each(func(name string, a *Allocation) {
  1353. a.reconcileNodes(nodeByProviderID)
  1354. a.reconcileDisks(diskByName)
  1355. })
  1356. return nil
  1357. }
  1358. func (a *Allocation) reconcileNodes(nodeByProviderID map[string]*Node) {
  1359. providerId := a.Properties.ProviderID
  1360. // Reconcile with node Assets
  1361. node, ok := nodeByProviderID[providerId]
  1362. if !ok {
  1363. // Failed to find node for allocation
  1364. return
  1365. }
  1366. // adjustmentRate is used to scale resource costs proportionally
  1367. // by the adjustment. This is necessary because we only get one
  1368. // adjustment per Node, not one per-resource-per-Node.
  1369. //
  1370. // e.g. total cost = $90, adjustment = -$10 => 0.9
  1371. // e.g. total cost = $150, adjustment = -$300 => 0.3333
  1372. // e.g. total cost = $150, adjustment = $50 => 1.5
  1373. adjustmentRate := 1.0
  1374. if node.TotalCost()-node.Adjustment() == 0 {
  1375. // If (totalCost - adjustment) is 0.0 then adjustment cancels
  1376. // the entire node cost and we should make everything 0
  1377. // without dividing by 0.
  1378. adjustmentRate = 0.0
  1379. } else if node.Adjustment() != 0.0 {
  1380. // adjustmentRate is the ratio of cost-with-adjustment (i.e. TotalCost)
  1381. // to cost-without-adjustment (i.e. TotalCost - Adjustment).
  1382. adjustmentRate = node.TotalCost() / (node.TotalCost() - node.Adjustment())
  1383. }
  1384. // Find total cost of each node resource for the window
  1385. cpuCost := node.CPUCost * (1.0 - node.Discount) * adjustmentRate
  1386. ramCost := node.RAMCost * (1.0 - node.Discount) * adjustmentRate
  1387. gpuCost := node.GPUCost * adjustmentRate
  1388. // Find the proportion of resource hours used by the allocation, checking for 0 denominators
  1389. cpuUsageProportion := 0.0
  1390. if node.CPUCoreHours != 0 {
  1391. cpuUsageProportion = a.CPUCoreHours / node.CPUCoreHours
  1392. } else {
  1393. log.Warningf("Missing CPU Hours for node Provider ID: %s", providerId)
  1394. }
  1395. ramUsageProportion := 0.0
  1396. if node.RAMByteHours != 0 {
  1397. ramUsageProportion = a.RAMByteHours / node.RAMByteHours
  1398. } else {
  1399. log.Warningf("Missing Ram Byte Hours for node Provider ID: %s", providerId)
  1400. }
  1401. gpuUsageProportion := 0.0
  1402. if node.GPUHours != 0 {
  1403. gpuUsageProportion = a.GPUHours / node.GPUHours
  1404. }
  1405. // No log for GPU because not all nodes have GPU
  1406. // Calculate the allocation's resource costs by the proportion of resources used and total costs
  1407. allocCPUCost := cpuUsageProportion * cpuCost
  1408. allocRAMCost := ramUsageProportion * ramCost
  1409. allocGPUCost := gpuUsageProportion * gpuCost
  1410. a.CPUCostAdjustment = allocCPUCost - a.CPUCost
  1411. a.RAMCostAdjustment = allocRAMCost - a.RAMCost
  1412. a.GPUCostAdjustment = allocGPUCost - a.GPUCost
  1413. }
  1414. func (a *Allocation) reconcileDisks(diskByName map[string]*Disk) {
  1415. pvBreakdown := a.Properties.PVBreakdown
  1416. if pvBreakdown == nil {
  1417. // No PV usage to reconcile
  1418. return
  1419. }
  1420. // Set PV Adjustment for allocation to 0 for idempotency
  1421. a.PVCostAdjustment = 0.0
  1422. for pvName, pvUsage := range pvBreakdown {
  1423. disk, ok := diskByName[pvName]
  1424. if !ok {
  1425. // Failed to find disk in assets
  1426. continue
  1427. }
  1428. // Check the proportion of disk that is being used by
  1429. pvUsageProportion := 0.0
  1430. if disk.ByteHours != 0 {
  1431. pvUsageProportion = pvUsage.ByteHours / disk.ByteHours
  1432. } else {
  1433. log.Warningf("Missing Byte Hours for disk: %s", pvName)
  1434. }
  1435. // take proportion of disk adjusted cost
  1436. allocPVCost := pvUsageProportion * disk.TotalCost()
  1437. // PVCostAdjustment is cumulative as there can be many PVs for each Allocation
  1438. a.PVCostAdjustment += allocPVCost - pvUsage.Cost
  1439. }
  1440. }
  1441. // Delete removes the allocation with the given name from the set
  1442. func (as *AllocationSet) Delete(name string) {
  1443. if as == nil {
  1444. return
  1445. }
  1446. as.Lock()
  1447. defer as.Unlock()
  1448. delete(as.externalKeys, name)
  1449. delete(as.idleKeys, name)
  1450. delete(as.allocations, name)
  1451. }
  1452. // Each invokes the given function for each Allocation in the set
  1453. func (as *AllocationSet) Each(f func(string, *Allocation)) {
  1454. if as == nil {
  1455. return
  1456. }
  1457. for k, a := range as.allocations {
  1458. f(k, a)
  1459. }
  1460. }
  1461. // End returns the End time of the AllocationSet window
  1462. func (as *AllocationSet) End() time.Time {
  1463. if as == nil {
  1464. log.Warningf("Allocation ETL: calling End on nil AllocationSet")
  1465. return time.Unix(0, 0)
  1466. }
  1467. if as.Window.End() == nil {
  1468. log.Warningf("Allocation ETL: AllocationSet with illegal window: End is nil; len(as.allocations)=%d", len(as.allocations))
  1469. return time.Unix(0, 0)
  1470. }
  1471. return *as.Window.End()
  1472. }
  1473. // Get returns the Allocation at the given key in the AllocationSet
  1474. func (as *AllocationSet) Get(key string) *Allocation {
  1475. as.RLock()
  1476. defer as.RUnlock()
  1477. if alloc, ok := as.allocations[key]; ok {
  1478. return alloc
  1479. }
  1480. return nil
  1481. }
  1482. // ExternalAllocations returns a map of the external allocations in the set.
  1483. // Returns clones of the actual Allocations, so mutability is not a problem.
  1484. func (as *AllocationSet) ExternalAllocations() map[string]*Allocation {
  1485. externals := map[string]*Allocation{}
  1486. if as.IsEmpty() {
  1487. return externals
  1488. }
  1489. as.RLock()
  1490. defer as.RUnlock()
  1491. for key := range as.externalKeys {
  1492. if alloc, ok := as.allocations[key]; ok {
  1493. externals[key] = alloc.Clone()
  1494. }
  1495. }
  1496. return externals
  1497. }
  1498. // ExternalCost returns the total aggregated external costs of the set
  1499. func (as *AllocationSet) ExternalCost() float64 {
  1500. if as.IsEmpty() {
  1501. return 0.0
  1502. }
  1503. as.RLock()
  1504. defer as.RUnlock()
  1505. externalCost := 0.0
  1506. for _, alloc := range as.allocations {
  1507. externalCost += alloc.ExternalCost
  1508. }
  1509. return externalCost
  1510. }
  1511. // IdleAllocations returns a map of the idle allocations in the AllocationSet.
  1512. // Returns clones of the actual Allocations, so mutability is not a problem.
  1513. func (as *AllocationSet) IdleAllocations() map[string]*Allocation {
  1514. idles := map[string]*Allocation{}
  1515. if as.IsEmpty() {
  1516. return idles
  1517. }
  1518. as.RLock()
  1519. defer as.RUnlock()
  1520. for key := range as.idleKeys {
  1521. if alloc, ok := as.allocations[key]; ok {
  1522. idles[key] = alloc.Clone()
  1523. }
  1524. }
  1525. return idles
  1526. }
  1527. // Insert aggregates the current entry in the AllocationSet by the given Allocation,
  1528. // but only if the Allocation is valid, i.e. matches the AllocationSet's window. If
  1529. // there is no existing entry, one is created. Nil error response indicates success.
  1530. func (as *AllocationSet) Insert(that *Allocation) error {
  1531. return as.insert(that)
  1532. }
  1533. func (as *AllocationSet) insert(that *Allocation) error {
  1534. if as == nil {
  1535. return fmt.Errorf("cannot insert into nil AllocationSet")
  1536. }
  1537. as.Lock()
  1538. defer as.Unlock()
  1539. if as.allocations == nil {
  1540. as.allocations = map[string]*Allocation{}
  1541. }
  1542. if as.externalKeys == nil {
  1543. as.externalKeys = map[string]bool{}
  1544. }
  1545. if as.idleKeys == nil {
  1546. as.idleKeys = map[string]bool{}
  1547. }
  1548. // Add the given Allocation to the existing entry, if there is one;
  1549. // otherwise just set directly into allocations
  1550. if _, ok := as.allocations[that.Name]; !ok {
  1551. as.allocations[that.Name] = that
  1552. } else {
  1553. as.allocations[that.Name].add(that)
  1554. }
  1555. // If the given Allocation is an external one, record that
  1556. if that.IsExternal() {
  1557. as.externalKeys[that.Name] = true
  1558. }
  1559. // If the given Allocation is an idle one, record that
  1560. if that.IsIdle() {
  1561. as.idleKeys[that.Name] = true
  1562. }
  1563. return nil
  1564. }
  1565. // IsEmpty returns true if the AllocationSet is nil, or if it contains
  1566. // zero allocations.
  1567. func (as *AllocationSet) IsEmpty() bool {
  1568. if as == nil || len(as.allocations) == 0 {
  1569. return true
  1570. }
  1571. as.RLock()
  1572. defer as.RUnlock()
  1573. return as.allocations == nil || len(as.allocations) == 0
  1574. }
  1575. // Length returns the number of Allocations in the set
  1576. func (as *AllocationSet) Length() int {
  1577. if as == nil {
  1578. return 0
  1579. }
  1580. as.RLock()
  1581. defer as.RUnlock()
  1582. return len(as.allocations)
  1583. }
  1584. // Map clones and returns a map of the AllocationSet's Allocations
  1585. func (as *AllocationSet) Map() map[string]*Allocation {
  1586. if as.IsEmpty() {
  1587. return map[string]*Allocation{}
  1588. }
  1589. return as.Clone().allocations
  1590. }
  1591. // MarshalJSON JSON-encodes the AllocationSet
  1592. func (as *AllocationSet) MarshalJSON() ([]byte, error) {
  1593. as.RLock()
  1594. defer as.RUnlock()
  1595. return json.Marshal(as.allocations)
  1596. }
  1597. // Resolution returns the AllocationSet's window duration
  1598. func (as *AllocationSet) Resolution() time.Duration {
  1599. return as.Window.Duration()
  1600. }
  1601. // Set uses the given Allocation to overwrite the existing entry in the
  1602. // AllocationSet under the Allocation's name.
  1603. func (as *AllocationSet) Set(alloc *Allocation) error {
  1604. if as.IsEmpty() {
  1605. as.Lock()
  1606. as.allocations = map[string]*Allocation{}
  1607. as.externalKeys = map[string]bool{}
  1608. as.idleKeys = map[string]bool{}
  1609. as.Unlock()
  1610. }
  1611. as.Lock()
  1612. defer as.Unlock()
  1613. as.allocations[alloc.Name] = alloc
  1614. // If the given Allocation is an external one, record that
  1615. if alloc.IsExternal() {
  1616. as.externalKeys[alloc.Name] = true
  1617. }
  1618. // If the given Allocation is an idle one, record that
  1619. if alloc.IsIdle() {
  1620. as.idleKeys[alloc.Name] = true
  1621. }
  1622. return nil
  1623. }
  1624. // Start returns the Start time of the AllocationSet window
  1625. func (as *AllocationSet) Start() time.Time {
  1626. if as == nil {
  1627. log.Warningf("Allocation ETL: calling Start on nil AllocationSet")
  1628. return time.Unix(0, 0)
  1629. }
  1630. if as.Window.Start() == nil {
  1631. log.Warningf("Allocation ETL: AllocationSet with illegal window: Start is nil; len(as.allocations)=%d", len(as.allocations))
  1632. return time.Unix(0, 0)
  1633. }
  1634. return *as.Window.Start()
  1635. }
  1636. // String represents the given Allocation as a string
  1637. func (as *AllocationSet) String() string {
  1638. if as == nil {
  1639. return "<nil>"
  1640. }
  1641. return fmt.Sprintf("AllocationSet{length: %d; window: %s; totalCost: %.2f}",
  1642. as.Length(), as.Window, as.TotalCost())
  1643. }
  1644. // TotalCost returns the sum of all TotalCosts of the allocations contained
  1645. func (as *AllocationSet) TotalCost() float64 {
  1646. if as.IsEmpty() {
  1647. return 0.0
  1648. }
  1649. as.RLock()
  1650. defer as.RUnlock()
  1651. tc := 0.0
  1652. for _, a := range as.allocations {
  1653. tc += a.TotalCost()
  1654. }
  1655. return tc
  1656. }
  1657. // UTCOffset returns the AllocationSet's configured UTCOffset.
  1658. func (as *AllocationSet) UTCOffset() time.Duration {
  1659. _, zone := as.Start().Zone()
  1660. return time.Duration(zone) * time.Second
  1661. }
  1662. func (as *AllocationSet) accumulate(that *AllocationSet) (*AllocationSet, error) {
  1663. if as.IsEmpty() {
  1664. return that, nil
  1665. }
  1666. if that.IsEmpty() {
  1667. return as, nil
  1668. }
  1669. // Set start, end to min(start), max(end)
  1670. start := as.Start()
  1671. end := as.End()
  1672. if that.Start().Before(start) {
  1673. start = that.Start()
  1674. }
  1675. if that.End().After(end) {
  1676. end = that.End()
  1677. }
  1678. acc := NewAllocationSet(start, end)
  1679. as.RLock()
  1680. defer as.RUnlock()
  1681. that.RLock()
  1682. defer that.RUnlock()
  1683. for _, alloc := range as.allocations {
  1684. err := acc.insert(alloc)
  1685. if err != nil {
  1686. return nil, err
  1687. }
  1688. }
  1689. for _, alloc := range that.allocations {
  1690. err := acc.insert(alloc)
  1691. if err != nil {
  1692. return nil, err
  1693. }
  1694. }
  1695. return acc, nil
  1696. }
  1697. // AllocationSetRange is a thread-safe slice of AllocationSets. It is meant to
  1698. // be used such that the AllocationSets held are consecutive and coherent with
  1699. // respect to using the same aggregation properties, UTC offset, and
  1700. // resolution. However these rules are not necessarily enforced, so use wisely.
  1701. type AllocationSetRange struct {
  1702. sync.RWMutex
  1703. allocations []*AllocationSet
  1704. }
  1705. // NewAllocationSetRange instantiates a new range composed of the given
  1706. // AllocationSets in the order provided.
  1707. func NewAllocationSetRange(allocs ...*AllocationSet) *AllocationSetRange {
  1708. return &AllocationSetRange{
  1709. allocations: allocs,
  1710. }
  1711. }
  1712. // Accumulate sums each AllocationSet in the given range, returning a single cumulative
  1713. // AllocationSet for the entire range.
  1714. func (asr *AllocationSetRange) Accumulate() (*AllocationSet, error) {
  1715. var allocSet *AllocationSet
  1716. var err error
  1717. asr.RLock()
  1718. defer asr.RUnlock()
  1719. for _, as := range asr.allocations {
  1720. allocSet, err = allocSet.accumulate(as)
  1721. if err != nil {
  1722. return nil, err
  1723. }
  1724. }
  1725. return allocSet, nil
  1726. }
  1727. // TODO niko/etl accumulate into lower-resolution chunks of the given resolution
  1728. // func (asr *AllocationSetRange) AccumulateBy(resolution time.Duration) *AllocationSetRange
  1729. // AggregateBy aggregates each AllocationSet in the range by the given
  1730. // properties and options.
  1731. func (asr *AllocationSetRange) AggregateBy(aggregateBy []string, options *AllocationAggregationOptions) error {
  1732. aggRange := &AllocationSetRange{allocations: []*AllocationSet{}}
  1733. asr.Lock()
  1734. defer asr.Unlock()
  1735. for _, as := range asr.allocations {
  1736. err := as.AggregateBy(aggregateBy, options)
  1737. if err != nil {
  1738. return err
  1739. }
  1740. aggRange.allocations = append(aggRange.allocations, as)
  1741. }
  1742. asr.allocations = aggRange.allocations
  1743. return nil
  1744. }
  1745. // Append appends the given AllocationSet to the end of the range. It does not
  1746. // validate whether or not that violates window continuity.
  1747. func (asr *AllocationSetRange) Append(that *AllocationSet) {
  1748. asr.Lock()
  1749. defer asr.Unlock()
  1750. asr.allocations = append(asr.allocations, that)
  1751. }
  1752. // Each invokes the given function for each AllocationSet in the range
  1753. func (asr *AllocationSetRange) Each(f func(int, *AllocationSet)) {
  1754. if asr == nil {
  1755. return
  1756. }
  1757. for i, as := range asr.allocations {
  1758. f(i, as)
  1759. }
  1760. }
  1761. // Get retrieves the AllocationSet at the given index of the range.
  1762. func (asr *AllocationSetRange) Get(i int) (*AllocationSet, error) {
  1763. if i < 0 || i >= len(asr.allocations) {
  1764. return nil, fmt.Errorf("AllocationSetRange: index out of range: %d", i)
  1765. }
  1766. asr.RLock()
  1767. defer asr.RUnlock()
  1768. return asr.allocations[i], nil
  1769. }
  1770. // InsertRange merges the given AllocationSetRange into the receiving one by
  1771. // lining up sets with matching windows, then inserting each allocation from
  1772. // the given ASR into the respective set in the receiving ASR. If the given
  1773. // ASR contains an AllocationSet from a window that does not exist in the
  1774. // receiving ASR, then an error is returned. However, the given ASR does not
  1775. // need to cover the full range of the receiver.
  1776. func (asr *AllocationSetRange) InsertRange(that *AllocationSetRange) error {
  1777. if asr == nil {
  1778. return fmt.Errorf("cannot insert range into nil AllocationSetRange")
  1779. }
  1780. // keys maps window to index in asr
  1781. keys := map[string]int{}
  1782. asr.Each(func(i int, as *AllocationSet) {
  1783. if as == nil {
  1784. return
  1785. }
  1786. keys[as.Window.String()] = i
  1787. })
  1788. // Nothing to merge, so simply return
  1789. if len(keys) == 0 {
  1790. return nil
  1791. }
  1792. var err error
  1793. that.Each(func(j int, thatAS *AllocationSet) {
  1794. if thatAS == nil || err != nil {
  1795. return
  1796. }
  1797. // Find matching AllocationSet in asr
  1798. i, ok := keys[thatAS.Window.String()]
  1799. if !ok {
  1800. err = fmt.Errorf("cannot merge AllocationSet into window that does not exist: %s", thatAS.Window.String())
  1801. return
  1802. }
  1803. as, err := asr.Get(i)
  1804. if err != nil {
  1805. err = fmt.Errorf("AllocationSetRange index does not exist: %d", i)
  1806. return
  1807. }
  1808. // Insert each Allocation from the given set
  1809. thatAS.Each(func(k string, alloc *Allocation) {
  1810. err = as.Insert(alloc)
  1811. if err != nil {
  1812. err = fmt.Errorf("error inserting allocation: %s", err)
  1813. return
  1814. }
  1815. })
  1816. })
  1817. // err might be nil
  1818. return err
  1819. }
  1820. // Length returns the length of the range, which is zero if nil
  1821. func (asr *AllocationSetRange) Length() int {
  1822. if asr == nil || asr.allocations == nil {
  1823. return 0
  1824. }
  1825. asr.RLock()
  1826. defer asr.RUnlock()
  1827. return len(asr.allocations)
  1828. }
  1829. // MarshalJSON JSON-encodes the range
  1830. func (asr *AllocationSetRange) MarshalJSON() ([]byte, error) {
  1831. asr.RLock()
  1832. asr.RUnlock()
  1833. return json.Marshal(asr.allocations)
  1834. }
  1835. // Slice copies the underlying slice of AllocationSets, maintaining order,
  1836. // and returns the copied slice.
  1837. func (asr *AllocationSetRange) Slice() []*AllocationSet {
  1838. if asr == nil || asr.allocations == nil {
  1839. return nil
  1840. }
  1841. asr.RLock()
  1842. defer asr.RUnlock()
  1843. copy := []*AllocationSet{}
  1844. for _, as := range asr.allocations {
  1845. copy = append(copy, as.Clone())
  1846. }
  1847. return copy
  1848. }
  1849. // String represents the given AllocationSetRange as a string
  1850. func (asr *AllocationSetRange) String() string {
  1851. if asr == nil {
  1852. return "<nil>"
  1853. }
  1854. return fmt.Sprintf("AllocationSetRange{length: %d}", asr.Length())
  1855. }
  1856. // UTCOffset returns the detected UTCOffset of the AllocationSets within the
  1857. // range. Defaults to 0 if the range is nil or empty. Does not warn if there
  1858. // are sets with conflicting UTCOffsets (just returns the first).
  1859. func (asr *AllocationSetRange) UTCOffset() time.Duration {
  1860. if asr.Length() == 0 {
  1861. return 0
  1862. }
  1863. as, err := asr.Get(0)
  1864. if err != nil {
  1865. return 0
  1866. }
  1867. return as.UTCOffset()
  1868. }
  1869. // Window returns the full window that the AllocationSetRange spans, from the
  1870. // start of the first AllocationSet to the end of the last one.
  1871. func (asr *AllocationSetRange) Window() Window {
  1872. if asr == nil || asr.Length() == 0 {
  1873. return NewWindow(nil, nil)
  1874. }
  1875. start := asr.allocations[0].Start()
  1876. end := asr.allocations[asr.Length()-1].End()
  1877. return NewWindow(&start, &end)
  1878. }