allocation.go 68 KB

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