allocation.go 54 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838
  1. package kubecost
  2. import (
  3. "bytes"
  4. "encoding/json"
  5. "fmt"
  6. "sort"
  7. "strings"
  8. "sync"
  9. "time"
  10. "github.com/kubecost/cost-model/pkg/log"
  11. )
  12. // TODO Clean-up use of IsEmpty; nil checks should be separated for safety.
  13. // TODO Consider making Allocation an interface, which is fulfilled by structs
  14. // like KubernetesAllocation, IdleAllocation, and ExternalAllocation.
  15. // ExternalSuffix indicates an external allocation
  16. const ExternalSuffix = "__external__"
  17. // IdleSuffix indicates an idle allocation property
  18. const IdleSuffix = "__idle__"
  19. // SharedSuffix indicates an shared allocation property
  20. const SharedSuffix = "__shared__"
  21. // UnallocatedSuffix indicates an unallocated allocation property
  22. const UnallocatedSuffix = "__unallocated__"
  23. // ShareWeighted indicates that a shared resource should be shared as a
  24. // proportion of the cost of the remaining allocations.
  25. const ShareWeighted = "__weighted__"
  26. // ShareEven indicates that a shared resource should be shared evenly across
  27. // all remaining allocations.
  28. const ShareEven = "__even__"
  29. // ShareNone indicates that a shareable resource should not be shared
  30. const ShareNone = "__none__"
  31. // Allocation is a unit of resource allocation and cost for a given window
  32. // of time and for a given kubernetes construct with its associated set of
  33. // properties.
  34. // TODO niko/cdmr compute efficiency on the fly?
  35. type Allocation struct {
  36. Name string `json:"name"`
  37. Properties Properties `json:"properties,omitempty"`
  38. Window Window `json:"window"`
  39. Start time.Time `json:"start"`
  40. End time.Time `json:"end"`
  41. CPUCoreHours float64 `json:"cpuCoreHours"`
  42. CPUCoreRequestAverage float64 `json:"cpuCoreRequestAverage"`
  43. CPUCoreUsageAverage float64 `json:"cpuCoreUsageAverage"`
  44. CPUCost float64 `json:"cpuCost"`
  45. CPUEfficiency float64 `json:"cpuEfficiency"`
  46. GPUHours float64 `json:"gpuHours"`
  47. GPUCost float64 `json:"gpuCost"`
  48. NetworkCost float64 `json:"networkCost"`
  49. PVByteHours float64 `json:"pvByteHours"`
  50. PVCost float64 `json:"pvCost"`
  51. RAMByteHours float64 `json:"ramByteHours"`
  52. RAMBytesRequestAverage float64 `json:"ramBytesRequestAverage"`
  53. RAMBytesUsageAverage float64 `json:"ramBytesUsageAverage"`
  54. RAMCost float64 `json:"ramCost"`
  55. RAMEfficiency float64 `json:"ramEfficiency"`
  56. SharedCost float64 `json:"sharedCost"`
  57. ExternalCost float64 `json:"externalCost"`
  58. TotalCost float64 `json:"totalCost"`
  59. TotalEfficiency float64 `json:"totalEfficiency"`
  60. }
  61. // AllocationMatchFunc is a function that can be used to match Allocations by
  62. // returning true for any given Allocation if a condition is met.
  63. type AllocationMatchFunc func(*Allocation) bool
  64. // Add returns the result of summing the two given Allocations, which sums the
  65. // summary fields (e.g. costs, resources) and recomputes efficiency. Neither of
  66. // the two original Allocations are mutated in the process.
  67. func (a *Allocation) Add(that *Allocation) (*Allocation, error) {
  68. if a == nil {
  69. return that.Clone(), nil
  70. }
  71. if that == nil {
  72. return a.Clone(), nil
  73. }
  74. // Note: no need to clone "that", as add only mutates the receiver
  75. agg := a.Clone()
  76. agg.add(that)
  77. return agg, nil
  78. }
  79. // Clone returns a deep copy of the given Allocation
  80. func (a *Allocation) Clone() *Allocation {
  81. if a == nil {
  82. return nil
  83. }
  84. return &Allocation{
  85. Name: a.Name,
  86. Properties: a.Properties.Clone(),
  87. Window: a.Window.Clone(),
  88. Start: a.Start,
  89. End: a.End,
  90. CPUCoreHours: a.CPUCoreHours,
  91. CPUCost: a.CPUCost,
  92. CPUEfficiency: a.CPUEfficiency,
  93. GPUHours: a.GPUHours,
  94. GPUCost: a.GPUCost,
  95. NetworkCost: a.NetworkCost,
  96. PVByteHours: a.PVByteHours,
  97. PVCost: a.PVCost,
  98. RAMByteHours: a.RAMByteHours,
  99. RAMCost: a.RAMCost,
  100. RAMEfficiency: a.RAMEfficiency,
  101. SharedCost: a.SharedCost,
  102. ExternalCost: a.ExternalCost,
  103. TotalCost: a.TotalCost,
  104. TotalEfficiency: a.TotalEfficiency,
  105. }
  106. }
  107. // Equal returns true if the values held in the given Allocation precisely
  108. // match those of the receiving Allocation. nil does not match nil.
  109. func (a *Allocation) Equal(that *Allocation) bool {
  110. if a == nil || that == nil {
  111. return false
  112. }
  113. if a.Name != that.Name {
  114. return false
  115. }
  116. if !a.Properties.Equal(&that.Properties) {
  117. return false
  118. }
  119. if !a.Window.Equal(that.Window) {
  120. return false
  121. }
  122. if !a.Start.Equal(that.Start) {
  123. return false
  124. }
  125. if !a.End.Equal(that.End) {
  126. return false
  127. }
  128. if a.CPUCoreHours != that.CPUCoreHours {
  129. return false
  130. }
  131. if a.CPUCost != that.CPUCost {
  132. return false
  133. }
  134. if a.CPUEfficiency != that.CPUEfficiency {
  135. return false
  136. }
  137. if a.GPUHours != that.GPUHours {
  138. return false
  139. }
  140. if a.GPUCost != that.GPUCost {
  141. return false
  142. }
  143. if a.NetworkCost != that.NetworkCost {
  144. return false
  145. }
  146. if a.PVByteHours != that.PVByteHours {
  147. return false
  148. }
  149. if a.PVCost != that.PVCost {
  150. return false
  151. }
  152. if a.RAMByteHours != that.RAMByteHours {
  153. return false
  154. }
  155. if a.RAMCost != that.RAMCost {
  156. return false
  157. }
  158. if a.RAMEfficiency != that.RAMEfficiency {
  159. return false
  160. }
  161. if a.SharedCost != that.SharedCost {
  162. return false
  163. }
  164. if a.ExternalCost != that.ExternalCost {
  165. return false
  166. }
  167. if a.TotalCost != that.TotalCost {
  168. return false
  169. }
  170. if a.TotalEfficiency != that.TotalEfficiency {
  171. return false
  172. }
  173. return true
  174. }
  175. // CPUCores converts the Allocation's CPUCoreHours into average CPUCores
  176. func (a *Allocation) CPUCores() float64 {
  177. if a.Minutes() <= 0.0 {
  178. return 0.0
  179. }
  180. return a.CPUCoreHours / (a.Minutes() / 60.0)
  181. }
  182. // RAMBytes converts the Allocation's RAMByteHours into average RAMBytes
  183. func (a *Allocation) RAMBytes() float64 {
  184. if a.Minutes() <= 0.0 {
  185. return 0.0
  186. }
  187. return a.RAMByteHours / (a.Minutes() / 60.0)
  188. }
  189. // PVBytes converts the Allocation's PVByteHours into average PVBytes
  190. func (a *Allocation) PVBytes() float64 {
  191. if a.Minutes() <= 0.0 {
  192. return 0.0
  193. }
  194. return a.PVByteHours / (a.Minutes() / 60.0)
  195. }
  196. // MarshalJSON implements json.Marshal interface
  197. func (a *Allocation) MarshalJSON() ([]byte, error) {
  198. buffer := bytes.NewBufferString("{")
  199. jsonEncodeString(buffer, "name", a.Name, ",")
  200. jsonEncode(buffer, "properties", a.Properties, ",")
  201. jsonEncode(buffer, "window", a.Window, ",")
  202. jsonEncodeString(buffer, "start", a.Start.Format(timeFmt), ",")
  203. jsonEncodeString(buffer, "end", a.End.Format(timeFmt), ",")
  204. jsonEncodeFloat64(buffer, "minutes", a.Minutes(), ",")
  205. jsonEncodeFloat64(buffer, "cpuCores", a.CPUCores(), ",")
  206. jsonEncodeFloat64(buffer, "cpuCoreRequestAverage", a.CPUCoreRequestAverage, ",")
  207. jsonEncodeFloat64(buffer, "cpuCoreUsageAverage", a.CPUCoreUsageAverage, ",")
  208. jsonEncodeFloat64(buffer, "cpuCoreHours", a.CPUCoreHours, ",")
  209. jsonEncodeFloat64(buffer, "cpuCost", a.CPUCost, ",")
  210. jsonEncodeFloat64(buffer, "cpuEfficiency", a.CPUEfficiency, ",")
  211. jsonEncodeFloat64(buffer, "gpuHours", a.GPUHours, ",")
  212. jsonEncodeFloat64(buffer, "gpuCost", a.GPUCost, ",")
  213. jsonEncodeFloat64(buffer, "networkCost", a.NetworkCost, ",")
  214. jsonEncodeFloat64(buffer, "pvBytes", a.PVBytes(), ",")
  215. jsonEncodeFloat64(buffer, "pvByteHours", a.PVByteHours, ",")
  216. jsonEncodeFloat64(buffer, "pvCost", a.PVCost, ",")
  217. jsonEncodeFloat64(buffer, "ramBytes", a.RAMBytes(), ",")
  218. jsonEncodeFloat64(buffer, "ramByteRequestAverage", a.RAMBytesRequestAverage, ",")
  219. jsonEncodeFloat64(buffer, "ramByteUsageAverage", a.RAMBytesUsageAverage, ",")
  220. jsonEncodeFloat64(buffer, "ramByteHours", a.RAMByteHours, ",")
  221. jsonEncodeFloat64(buffer, "ramCost", a.RAMCost, ",")
  222. jsonEncodeFloat64(buffer, "ramEfficiency", a.RAMEfficiency, ",")
  223. jsonEncodeFloat64(buffer, "sharedCost", a.SharedCost, ",")
  224. jsonEncodeFloat64(buffer, "totalCost", a.TotalCost, ",")
  225. jsonEncodeFloat64(buffer, "totalEfficiency", a.TotalEfficiency, "")
  226. buffer.WriteString("}")
  227. return buffer.Bytes(), nil
  228. }
  229. // TODO niko/cdmr
  230. // func (a *Allocation)UnmarshalJSON()
  231. // Resolution returns the duration of time covered by the Allocation
  232. func (a *Allocation) Resolution() time.Duration {
  233. return a.End.Sub(a.Start)
  234. }
  235. // IsAggregated is true if the given Allocation has been aggregated, which we
  236. // define by a lack of Properties.
  237. func (a *Allocation) IsAggregated() bool {
  238. return a == nil || a.Properties == nil
  239. }
  240. // IsExternal is true if the given Allocation represents external costs.
  241. func (a *Allocation) IsExternal() bool {
  242. return strings.Contains(a.Name, ExternalSuffix)
  243. }
  244. // IsIdle is true if the given Allocation represents idle costs.
  245. func (a *Allocation) IsIdle() bool {
  246. return strings.Contains(a.Name, IdleSuffix)
  247. }
  248. // IsUnallocated is true if the given Allocation represents unallocated costs.
  249. func (a *Allocation) IsUnallocated() bool {
  250. return strings.Contains(a.Name, UnallocatedSuffix)
  251. }
  252. // Minutes returns the number of minutes the Allocation represents, as defined
  253. // by the difference between the end and start times.
  254. func (a *Allocation) Minutes() float64 {
  255. return a.End.Sub(a.Start).Minutes()
  256. }
  257. // Share works like Add, but converts the entire cost of the given Allocation
  258. // to SharedCost, rather than adding to the individual resource costs.
  259. // TODO niko/cdmr unit test changes!!!
  260. func (a *Allocation) Share(that *Allocation) (*Allocation, error) {
  261. if that == nil {
  262. return a.Clone(), nil
  263. }
  264. // Convert all costs of shared Allocation to SharedCost, zero out all
  265. // non-shared costs, then add.
  266. share := that.Clone()
  267. share.SharedCost += share.TotalCost
  268. share.TotalEfficiency = 1.0
  269. share.CPUCost = 0
  270. share.CPUCoreHours = 0
  271. share.CPUEfficiency = 0
  272. share.RAMCost = 0
  273. share.RAMByteHours = 0
  274. share.RAMEfficiency = 0
  275. share.GPUCost = 0
  276. share.GPUHours = 0
  277. share.PVCost = 0
  278. share.PVByteHours = 0
  279. share.NetworkCost = 0
  280. share.ExternalCost = 0
  281. if a == nil {
  282. return share, nil
  283. }
  284. agg := a.Clone()
  285. agg.add(that)
  286. return agg, nil
  287. }
  288. // String represents the given Allocation as a string
  289. func (a *Allocation) String() string {
  290. return fmt.Sprintf("%s%s=%.2f", a.Name, NewWindow(&a.Start, &a.End), a.TotalCost)
  291. }
  292. func (a *Allocation) add(that *Allocation) {
  293. if a == nil {
  294. log.Warningf("Allocation.AggregateBy: trying to add a nil receiver")
  295. return
  296. }
  297. aCluster, _ := a.Properties.GetCluster()
  298. thatCluster, _ := that.Properties.GetCluster()
  299. aNode, _ := a.Properties.GetNode()
  300. thatNode, _ := that.Properties.GetNode()
  301. // reset properties
  302. a.Properties = nil
  303. // ensure that we carry cluster ID and/or node over if they're the same
  304. // required for idle/shared cost allocation
  305. if aCluster == thatCluster {
  306. a.Properties = Properties{ClusterProp: aCluster}
  307. }
  308. if aNode == thatNode {
  309. if a.Properties == nil {
  310. a.Properties = Properties{NodeProp: aNode}
  311. } else {
  312. a.Properties.SetNode(aNode)
  313. }
  314. }
  315. // Expand Window, Start, and End to be the "max" of each between the two
  316. // given Allocations.
  317. a.Window = a.Window.Expand(that.Window)
  318. if that.Start.Before(a.Start) {
  319. a.Start = that.Start
  320. }
  321. if that.End.After(a.End) {
  322. a.End = that.End
  323. }
  324. // Note: efficiency numbers are computed the cost-weighted sum of each
  325. // Allocation's efficiency.
  326. // e.g. ($10 @ 25%) + ($10 @ 75%) = (2.5+7.5)/20 = 50%
  327. // e.g. ($90 @ 10%) + ($10 @ 100%) = (9.0+10.0)/100 = 19%
  328. // e.g. ($100 @ 0%) + ($100 @ 0%) = (0.0+0.0)/200 = 0%
  329. // e.g. ($10 @ 150%) + ($10 @ 50%) = (15.0+5.0)/20 = 100%
  330. // e.g. ($0 @ 100%) + ($0 @ 50%) = 0% (no div by 0)
  331. // Compute CPU efficiency (see note above for methodology)
  332. aggCPUCost := a.CPUCost + that.CPUCost
  333. if aggCPUCost > 0 {
  334. a.CPUEfficiency = (a.CPUEfficiency*a.CPUCost + that.CPUEfficiency*that.CPUCost) / aggCPUCost
  335. } else {
  336. a.CPUEfficiency = 0.0
  337. }
  338. // Compute RAM efficiency (see note above for methodology)
  339. aggRAMCost := a.RAMCost + that.RAMCost
  340. if aggRAMCost > 0 {
  341. a.RAMEfficiency = (a.RAMEfficiency*a.RAMCost + that.RAMEfficiency*that.RAMCost) / aggRAMCost
  342. } else {
  343. a.RAMEfficiency = 0.0
  344. }
  345. // Compute total efficiency (see note above for methodology)
  346. aggTotalCost := a.TotalCost + that.TotalCost
  347. if aggTotalCost > 0 {
  348. a.TotalEfficiency = (a.TotalEfficiency*a.TotalCost + that.TotalEfficiency*that.TotalCost) / aggTotalCost
  349. } else {
  350. aggTotalCost = 0.0
  351. }
  352. // Sum all cumulative resource fields
  353. a.CPUCoreHours += that.CPUCoreHours
  354. a.GPUHours += that.GPUHours
  355. a.RAMByteHours += that.RAMByteHours
  356. a.PVByteHours += that.PVByteHours
  357. // Sum all cumulative cost fields
  358. a.CPUCost += that.CPUCost
  359. a.GPUCost += that.GPUCost
  360. a.RAMCost += that.RAMCost
  361. a.PVCost += that.PVCost
  362. a.NetworkCost += that.NetworkCost
  363. a.SharedCost += that.SharedCost
  364. a.ExternalCost += that.ExternalCost
  365. a.TotalCost += that.TotalCost
  366. }
  367. // AllocationSet stores a set of Allocations, each with a unique name, that share
  368. // a window. An AllocationSet is mutable, so treat it like a threadsafe map.
  369. type AllocationSet struct {
  370. sync.RWMutex
  371. allocations map[string]*Allocation
  372. externalKeys map[string]bool
  373. idleKeys map[string]bool
  374. Window Window
  375. Warnings []string
  376. Errors []string
  377. }
  378. // NewAllocationSet instantiates a new AllocationSet and, optionally, inserts
  379. // the given list of Allocations
  380. func NewAllocationSet(start, end time.Time, allocs ...*Allocation) *AllocationSet {
  381. as := &AllocationSet{
  382. allocations: map[string]*Allocation{},
  383. externalKeys: map[string]bool{},
  384. idleKeys: map[string]bool{},
  385. Window: NewWindow(&start, &end),
  386. }
  387. for _, a := range allocs {
  388. as.Insert(a)
  389. }
  390. return as
  391. }
  392. // AllocationAggregationOptions provide advanced functionality to AggregateBy, including
  393. // filtering results and sharing allocations. FilterFuncs are a list of match
  394. // functions such that, if any function fails, the allocation is ignored.
  395. // ShareFuncs are a list of match functions such that, if any function
  396. // succeeds, the allocation is marked as a shared resource. ShareIdle is a
  397. // simple flag for sharing idle resources.
  398. type AllocationAggregationOptions struct {
  399. FilterFuncs []AllocationMatchFunc
  400. SplitIdle bool
  401. MergeUnallocated bool
  402. ShareFuncs []AllocationMatchFunc
  403. ShareIdle string
  404. ShareSplit string
  405. SharedHourlyCosts map[string]float64
  406. }
  407. // AggregateBy aggregates the Allocations in the given AllocationSet by the given
  408. // Property. This will only be legal if the AllocationSet is divisible by the
  409. // given Property; e.g. Containers can be divided by Namespace, but not vice-a-versa.
  410. func (as *AllocationSet) AggregateBy(properties Properties, options *AllocationAggregationOptions) error {
  411. // The order of operations for aggregating allocations is as follows:
  412. // 1. Partition external, idle, and shared allocations into separate sets
  413. // 2. Compute idle coefficients (if necessary)
  414. // a) if idle allocation is to be shared, compute idle coefficients
  415. // (do not compute shared coefficients here, see step 5)
  416. // b) if idle allocation is NOT shared, but filters are present, compute
  417. // idle filtration coefficients for the purpose of only returning the
  418. // portion of idle allocation that would have been shared with the
  419. // unfiltered results set. (See unit tests 5.a,b,c)
  420. // 3. Ignore allocation if it fails any of the FilterFuncs
  421. // 4. Distribute idle allocations among remaining non-idle, non-external
  422. // allocations
  423. // 5. Generate aggregation key and insert allocation into the output set
  424. // 6. Scale un-aggregated idle coefficients by filtration coefficient
  425. // 7. If there are shared allocations, compute sharing coefficients on
  426. // the aggregated set, then share allocation accordingly
  427. // 8. If there are external allocations that can be aggregated into
  428. // the output (i.e. they can be used to generate a valid key for
  429. // the given properties) then aggregate; otherwise... ignore them?
  430. // 9. If the merge idle option is enabled, merge any remaining idle
  431. // allocations into a single idle allocation
  432. // TODO niko/etl revisit (ShareIdle: ShareEven) case, which is probably wrong
  433. // (and, frankly, ill-defined; i.e. evenly across clusters? within clusters?)
  434. if options == nil {
  435. options = &AllocationAggregationOptions{}
  436. }
  437. if as.IsEmpty() {
  438. return nil
  439. }
  440. // aggSet will collect the aggregated allocations
  441. aggSet := &AllocationSet{
  442. Window: as.Window.Clone(),
  443. }
  444. // externalSet will collect external allocations
  445. externalSet := &AllocationSet{
  446. Window: as.Window.Clone(),
  447. }
  448. // idleSet will be shared among aggSet after initial aggregation
  449. // is complete
  450. idleSet := &AllocationSet{
  451. Window: as.Window.Clone(),
  452. }
  453. // shareSet will be shared among aggSet after initial aggregation
  454. // is complete
  455. shareSet := &AllocationSet{
  456. Window: as.Window.Clone(),
  457. }
  458. // Convert SharedHourlyCosts to Allocations in the shareSet
  459. for name, cost := range options.SharedHourlyCosts {
  460. if cost > 0.0 {
  461. hours := as.Resolution().Hours()
  462. // If set ends in the future, adjust hours accordingly
  463. diff := time.Now().Sub(as.End())
  464. if diff < 0.0 {
  465. hours += diff.Hours()
  466. }
  467. totalSharedCost := cost * hours
  468. shareSet.Insert(&Allocation{
  469. Name: fmt.Sprintf("%s/%s", name, SharedSuffix),
  470. Start: as.Start(),
  471. End: as.End(),
  472. SharedCost: totalSharedCost,
  473. TotalCost: totalSharedCost,
  474. })
  475. }
  476. }
  477. as.Lock()
  478. defer as.Unlock()
  479. // (1) Loop and find all of the external, idle, and shared allocations. Add
  480. // them to their respective sets, removing them from the set of allocations
  481. // to aggregate.
  482. for _, alloc := range as.allocations {
  483. // External allocations get aggregated post-hoc (see step 6) and do
  484. // not necessarily contain complete sets of properties, so they are
  485. // moved to a separate AllocationSet.
  486. if alloc.IsExternal() {
  487. delete(as.externalKeys, alloc.Name)
  488. delete(as.allocations, alloc.Name)
  489. externalSet.Insert(alloc)
  490. continue
  491. }
  492. cluster, err := alloc.Properties.GetCluster()
  493. if err != nil {
  494. log.Warningf("AllocationSet.AggregateBy: missing cluster for allocation: %s", alloc.Name)
  495. return err
  496. }
  497. // Idle allocations should be separated into idleSet if they are to be
  498. // shared later on. If they are not to be shared, then aggregate them.
  499. if alloc.IsIdle() {
  500. delete(as.idleKeys, alloc.Name)
  501. delete(as.allocations, alloc.Name)
  502. if options.ShareIdle == ShareEven || options.ShareIdle == ShareWeighted {
  503. idleSet.Insert(alloc)
  504. } else {
  505. aggSet.Insert(alloc)
  506. }
  507. continue
  508. }
  509. // Shared allocations must be identified and separated prior to
  510. // aggregation and filtering. That is, if any of the ShareFuncs
  511. // return true, then move the allocation to shareSet.
  512. for _, sf := range options.ShareFuncs {
  513. if sf(alloc) {
  514. delete(as.idleKeys, alloc.Name)
  515. delete(as.allocations, alloc.Name)
  516. alloc.Name = fmt.Sprintf("%s/%s", cluster, SharedSuffix)
  517. shareSet.Insert(alloc)
  518. break
  519. }
  520. }
  521. }
  522. // It's possible that no more un-shared, non-idle, non-external allocations
  523. // remain at this point. This always results in an emptySet.
  524. if len(as.allocations) == 0 {
  525. log.Warningf("ETL: AggregateBy: no allocations to aggregate")
  526. emptySet := &AllocationSet{
  527. Window: as.Window.Clone(),
  528. }
  529. as.allocations = emptySet.allocations
  530. return nil
  531. }
  532. // (2) In order to correctly apply idle and shared resource coefficients
  533. // appropriately, we need to determine the coefficients for the full set
  534. // of data. The ensures that the ratios are maintained through filtering.
  535. // idleCoefficients are organized by [cluster][allocation][resource]=coeff
  536. var idleCoefficients map[string]map[string]map[string]float64
  537. // shareCoefficients are organized by [allocation][resource]=coeff (no cluster)
  538. var shareCoefficients map[string]float64
  539. var err error
  540. // (2a) If there are idle costs and we intend to share them, compute the
  541. // coefficients for sharing the cost among the non-idle, non-aggregated
  542. // allocations.
  543. if idleSet.Length() > 0 && options.ShareIdle != ShareNone {
  544. idleCoefficients, err = computeIdleCoeffs(properties, options, as)
  545. if err != nil {
  546. log.Warningf("AllocationSet.AggregateBy: compute idle coeff: %s", err)
  547. return fmt.Errorf("error computing idle coefficients: %s", err)
  548. }
  549. }
  550. // (2b) If we're not sharing idle and we're filtering, we need to track the
  551. // amount of each idle allocation to "delete" in order to maintain parity
  552. // with the idle-allocated results. That is, we want to return only the
  553. // idle cost that would have been shared with the unfiltered portion of
  554. // the results, not the full idle cost.
  555. var idleFiltrationCoefficients map[string]map[string]map[string]float64
  556. if len(options.FilterFuncs) > 0 && options.ShareIdle == ShareNone {
  557. idleFiltrationCoefficients, err = computeIdleCoeffs(properties, options, as)
  558. if err != nil {
  559. log.Warningf("AllocationSet.AggregateBy: compute idle coeff: %s", err)
  560. return fmt.Errorf("error computing idle filtration coefficients: %s", err)
  561. }
  562. }
  563. // (3-5) Filter, distribute idle cost, and aggregate (in that order)
  564. for _, alloc := range as.allocations {
  565. cluster, err := alloc.Properties.GetCluster()
  566. if err != nil {
  567. log.Warningf("AllocationSet.AggregateBy: missing cluster for allocation: %s", alloc.Name)
  568. return err
  569. }
  570. skip := false
  571. // (3) If any of the filter funcs fail, immediately skip the allocation.
  572. for _, ff := range options.FilterFuncs {
  573. if !ff(alloc) {
  574. skip = true
  575. break
  576. }
  577. }
  578. if skip {
  579. // If we are tracking idle filtration coefficients, delete the
  580. // entry corresponding to the filtered allocation. (Deleting the
  581. // entry will result in that proportional amount being removed
  582. // from the idle allocation at the end of the process.)
  583. if idleFiltrationCoefficients != nil {
  584. if ifcc, ok := idleFiltrationCoefficients[cluster]; ok {
  585. delete(ifcc, alloc.Name)
  586. }
  587. }
  588. continue
  589. }
  590. // (4) Split idle allocations and distribute among remaining
  591. // un-aggregated allocations.
  592. // NOTE: if idle allocation is off (i.e. ShareIdle == ShareNone) then
  593. // all idle allocations will be in the aggSet at this point, so idleSet
  594. // will be empty and we won't enter this block.
  595. if idleSet.Length() > 0 {
  596. // Distribute idle allocations by coefficient per-cluster, per-allocation
  597. for _, idleAlloc := range idleSet.allocations {
  598. // Only share idle if the cluster matches; i.e. the allocation
  599. // is from the same cluster as the idle costs
  600. idleCluster, err := idleAlloc.Properties.GetCluster()
  601. if err != nil {
  602. return err
  603. }
  604. if idleCluster != cluster {
  605. continue
  606. }
  607. // Make sure idle coefficients exist
  608. if _, ok := idleCoefficients[cluster]; !ok {
  609. log.Errorf("ETL: share (idle) allocation: error getting allocation coefficient [no cluster: '%s' in coefficients] for '%s'", cluster, alloc.Name)
  610. continue
  611. }
  612. if _, ok := idleCoefficients[cluster][alloc.Name]; !ok {
  613. log.Errorf("ETL: share (idle) allocation: error getting allocation coefficienct for '%s'", alloc.Name)
  614. continue
  615. }
  616. alloc.CPUCoreHours += idleAlloc.CPUCoreHours * idleCoefficients[cluster][alloc.Name]["cpu"]
  617. alloc.GPUHours += idleAlloc.GPUHours * idleCoefficients[cluster][alloc.Name]["gpu"]
  618. alloc.RAMByteHours += idleAlloc.RAMByteHours * idleCoefficients[cluster][alloc.Name]["ram"]
  619. idleCPUCost := idleAlloc.CPUCost * idleCoefficients[cluster][alloc.Name]["cpu"]
  620. idleGPUCost := idleAlloc.GPUCost * idleCoefficients[cluster][alloc.Name]["gpu"]
  621. idleRAMCost := idleAlloc.RAMCost * idleCoefficients[cluster][alloc.Name]["ram"]
  622. alloc.CPUCost += idleCPUCost
  623. alloc.GPUCost += idleGPUCost
  624. alloc.RAMCost += idleRAMCost
  625. alloc.TotalCost += idleCPUCost + idleGPUCost + idleRAMCost
  626. }
  627. }
  628. // (5) generate key to use for aggregation-by-key and allocation name
  629. key, err := alloc.generateKey(properties)
  630. if err != nil {
  631. return err
  632. }
  633. alloc.Name = key
  634. if options.MergeUnallocated && alloc.IsUnallocated() {
  635. alloc.Name = UnallocatedSuffix
  636. }
  637. // Inserting the allocation with the generated key for a name will
  638. // perform the actual basic aggregation step.
  639. aggSet.Insert(alloc)
  640. }
  641. // clusterIdleFiltrationCoeffs is used to track per-resource idle
  642. // coefficients on a cluster-by-cluster basis. It is, essentailly, an
  643. // aggregation of idleFiltrationCoefficients after they have been
  644. // filtered above (in step 3)
  645. var clusterIdleFiltrationCoeffs map[string]map[string]float64
  646. if idleFiltrationCoefficients != nil {
  647. clusterIdleFiltrationCoeffs = map[string]map[string]float64{}
  648. for cluster, m := range idleFiltrationCoefficients {
  649. if _, ok := clusterIdleFiltrationCoeffs[cluster]; !ok {
  650. clusterIdleFiltrationCoeffs[cluster] = map[string]float64{
  651. "cpu": 0.0,
  652. "gpu": 0.0,
  653. "ram": 0.0,
  654. }
  655. }
  656. for _, n := range m {
  657. for resource, val := range n {
  658. clusterIdleFiltrationCoeffs[cluster][resource] += val
  659. }
  660. }
  661. }
  662. }
  663. // (6) If we have both un-shared idle allocations and idle filtration
  664. // coefficients (i.e. we have computed coefficients for scaling idle
  665. // allocation costs by cluster) then use those coefficients to scale down
  666. // each idle allocation.
  667. if len(aggSet.idleKeys) > 0 && clusterIdleFiltrationCoeffs != nil {
  668. for idleKey := range aggSet.idleKeys {
  669. idleAlloc := aggSet.Get(idleKey)
  670. cluster, err := idleAlloc.Properties.GetCluster()
  671. if err != nil {
  672. log.Warningf("AggregateBy: idle allocation without cluster: %s", idleAlloc)
  673. }
  674. if resourceCoeffs, ok := clusterIdleFiltrationCoeffs[cluster]; ok {
  675. idleAlloc.CPUCost *= resourceCoeffs["cpu"]
  676. idleAlloc.CPUCoreHours *= resourceCoeffs["cpu"]
  677. idleAlloc.RAMCost *= resourceCoeffs["ram"]
  678. idleAlloc.RAMByteHours *= resourceCoeffs["ram"]
  679. idleAlloc.TotalCost = idleAlloc.CPUCost + idleAlloc.RAMCost
  680. }
  681. }
  682. }
  683. // (7) Split shared allocations and distribute among aggregated allocations
  684. if shareSet.Length() > 0 {
  685. shareCoefficients, err = computeShareCoeffs(properties, options, aggSet)
  686. if err != nil {
  687. log.Warningf("AllocationSet.AggregateBy: compute shared coeff: missing cluster ID: %s", err)
  688. return err
  689. }
  690. for _, alloc := range aggSet.allocations {
  691. if alloc.IsIdle() {
  692. // Skip idle allocations (they do not receive shared allocation)
  693. continue
  694. }
  695. // Distribute shared allocations by coefficient per-allocation
  696. // NOTE: share coefficients do not partition by cluster, like
  697. // idle coefficients do.
  698. for _, sharedAlloc := range shareSet.allocations {
  699. if _, ok := shareCoefficients[alloc.Name]; !ok {
  700. log.Errorf("ETL: share allocation: error getting allocation coefficienct for '%s'", alloc.Name)
  701. continue
  702. }
  703. alloc.SharedCost += sharedAlloc.TotalCost * shareCoefficients[alloc.Name]
  704. alloc.TotalCost += sharedAlloc.TotalCost * shareCoefficients[alloc.Name]
  705. }
  706. }
  707. }
  708. // (8) Aggregate external allocations into aggregated allocations. This may
  709. // not be possible for every external allocation, but attempt to find an
  710. // exact key match, given each external allocation's proerties, and
  711. // aggregate if an exact match is found.
  712. for _, alloc := range externalSet.allocations {
  713. key, err := alloc.generateKey(properties)
  714. if err != nil {
  715. continue
  716. }
  717. alloc.Name = key
  718. aggSet.Insert(alloc)
  719. }
  720. // (9) Combine all idle allocations into a single "__idle__" allocation
  721. if !options.SplitIdle {
  722. for _, idleAlloc := range aggSet.IdleAllocations() {
  723. aggSet.Delete(idleAlloc.Name)
  724. idleAlloc.Name = IdleSuffix
  725. aggSet.Insert(idleAlloc)
  726. }
  727. }
  728. as.allocations = aggSet.allocations
  729. return nil
  730. }
  731. // TODO niko/etl deprecate the use of a map of resources here, we only use totals
  732. func computeShareCoeffs(properties Properties, options *AllocationAggregationOptions, as *AllocationSet) (map[string]float64, error) {
  733. // Compute coeffs by totalling per-allocation, then dividing by the total.
  734. coeffs := map[string]float64{}
  735. // Compute totals for all allocations
  736. total := 0.0
  737. // ShareEven counts each aggregation with even weight, whereas ShareWeighted
  738. // counts each aggregation proportionally to its respective costs
  739. shareType := options.ShareSplit
  740. // Record allocation values first, then normalize by totals to get percentages
  741. for name, alloc := range as.allocations {
  742. if alloc.IsIdle() {
  743. // Skip idle allocations in coefficient calculation
  744. continue
  745. }
  746. if shareType == ShareEven {
  747. // Not additive - set to 1.0 for even distribution
  748. coeffs[name] = 1.0
  749. // Total is always additive
  750. total += 1.0
  751. } else {
  752. // Both are additive for weighted distribution
  753. coeffs[name] += alloc.TotalCost
  754. total += alloc.TotalCost
  755. }
  756. }
  757. // Normalize coefficients by totals
  758. for a := range coeffs {
  759. if coeffs[a] > 0 && total > 0 {
  760. coeffs[a] /= total
  761. } else {
  762. log.Warningf("ETL: invalid values for shared coefficients: %d, %d", coeffs[a], total)
  763. coeffs[a] = 0.0
  764. }
  765. }
  766. return coeffs, nil
  767. }
  768. func computeIdleCoeffs(properties Properties, options *AllocationAggregationOptions, as *AllocationSet) (map[string]map[string]map[string]float64, error) {
  769. types := []string{"cpu", "gpu", "ram"}
  770. // Compute idle coefficients, then save them in AllocationAggregationOptions
  771. coeffs := map[string]map[string]map[string]float64{}
  772. // Compute totals per resource for CPU, GPU, RAM, and PV
  773. totals := map[string]map[string]float64{}
  774. // ShareEven counts each allocation with even weight, whereas ShareWeighted
  775. // counts each allocation proportionally to its respective costs
  776. shareType := options.ShareIdle
  777. // Record allocation values first, then normalize by totals to get percentages
  778. for _, alloc := range as.allocations {
  779. if alloc.IsIdle() {
  780. // Skip idle allocations in coefficient calculation
  781. continue
  782. }
  783. // If any of the share funcs succeed, share the allocation. Do this
  784. // prior to filtering so that shared namespaces, etc do not get
  785. // filtered out before we have a chance to share them.
  786. skip := false
  787. for _, sf := range options.ShareFuncs {
  788. if sf(alloc) {
  789. skip = true
  790. break
  791. }
  792. }
  793. if skip {
  794. continue
  795. }
  796. // We need to key the allocations by cluster id
  797. clusterID, err := alloc.Properties.GetCluster()
  798. if err != nil {
  799. return nil, err
  800. }
  801. // get the name key for the allocation
  802. name := alloc.Name
  803. // Create cluster based tables if they don't exist
  804. if _, ok := coeffs[clusterID]; !ok {
  805. coeffs[clusterID] = map[string]map[string]float64{}
  806. }
  807. if _, ok := totals[clusterID]; !ok {
  808. totals[clusterID] = map[string]float64{}
  809. }
  810. if _, ok := coeffs[clusterID][name]; !ok {
  811. coeffs[clusterID][name] = map[string]float64{}
  812. }
  813. if shareType == ShareEven {
  814. for _, r := range types {
  815. // Not additive - hard set to 1.0
  816. coeffs[clusterID][name][r] = 1.0
  817. // totals are additive
  818. totals[clusterID][r] += 1.0
  819. }
  820. } else {
  821. coeffs[clusterID][name]["cpu"] += alloc.CPUCost
  822. coeffs[clusterID][name]["gpu"] += alloc.GPUCost
  823. coeffs[clusterID][name]["ram"] += alloc.RAMCost
  824. totals[clusterID]["cpu"] += alloc.CPUCost
  825. totals[clusterID]["gpu"] += alloc.GPUCost
  826. totals[clusterID]["ram"] += alloc.RAMCost
  827. }
  828. }
  829. // Normalize coefficients by totals
  830. for c := range coeffs {
  831. for a := range coeffs[c] {
  832. for _, r := range types {
  833. if coeffs[c][a][r] > 0 && totals[c][r] > 0 {
  834. coeffs[c][a][r] /= totals[c][r]
  835. }
  836. }
  837. }
  838. }
  839. return coeffs, nil
  840. }
  841. func (alloc *Allocation) generateKey(properties Properties) (string, error) {
  842. // Names will ultimately be joined into a single name, which uniquely
  843. // identifies allocations.
  844. names := []string{}
  845. if properties.HasCluster() {
  846. cluster, err := alloc.Properties.GetCluster()
  847. if err != nil {
  848. return "", err
  849. }
  850. names = append(names, cluster)
  851. }
  852. if properties.HasNode() {
  853. node, err := alloc.Properties.GetNode()
  854. if err != nil {
  855. return "", err
  856. }
  857. names = append(names, node)
  858. }
  859. if properties.HasNamespace() {
  860. namespace, err := alloc.Properties.GetNamespace()
  861. if err != nil {
  862. return "", err
  863. }
  864. names = append(names, namespace)
  865. }
  866. if properties.HasControllerKind() {
  867. controllerKind, err := alloc.Properties.GetControllerKind()
  868. if err != nil {
  869. // Indicate that allocation has no controller
  870. controllerKind = UnallocatedSuffix
  871. }
  872. if prop, _ := properties.GetControllerKind(); prop != "" && prop != controllerKind {
  873. // The allocation does not have the specified controller kind
  874. controllerKind = UnallocatedSuffix
  875. }
  876. names = append(names, controllerKind)
  877. }
  878. if properties.HasController() {
  879. if !properties.HasControllerKind() {
  880. controllerKind, err := alloc.Properties.GetControllerKind()
  881. if err == nil {
  882. names = append(names, controllerKind)
  883. }
  884. }
  885. controller, err := alloc.Properties.GetController()
  886. if err != nil {
  887. // Indicate that allocation has no controller
  888. controller = UnallocatedSuffix
  889. }
  890. names = append(names, controller)
  891. }
  892. if properties.HasPod() {
  893. pod, err := alloc.Properties.GetPod()
  894. if err != nil {
  895. return "", err
  896. }
  897. names = append(names, pod)
  898. }
  899. if properties.HasContainer() {
  900. container, err := alloc.Properties.GetContainer()
  901. if err != nil {
  902. return "", err
  903. }
  904. names = append(names, container)
  905. }
  906. if properties.HasService() {
  907. services, err := alloc.Properties.GetServices()
  908. if err != nil {
  909. // Indicate that allocation has no services
  910. names = append(names, UnallocatedSuffix)
  911. } else {
  912. // TODO niko/etl support multi-service aggregation
  913. if len(services) > 0 {
  914. for _, service := range services {
  915. names = append(names, service)
  916. break
  917. }
  918. } else {
  919. // Indicate that allocation has no services
  920. names = append(names, UnallocatedSuffix)
  921. }
  922. }
  923. }
  924. if properties.HasAnnotations() {
  925. annotations, err := alloc.Properties.GetAnnotations() // annotations that the individual allocation possesses
  926. if err != nil {
  927. // Indicate that allocation has no annotations
  928. names = append(names, UnallocatedSuffix)
  929. } else {
  930. annotationNames := []string{}
  931. aggAnnotations, err := properties.GetAnnotations() // potential annotations to aggregate on supplied by the API caller
  932. if err != nil {
  933. // We've already checked HasAnnotation, so this should never occur
  934. return "", err
  935. }
  936. // calvin - support multi-annotation aggregation
  937. for annotationName := range aggAnnotations {
  938. if val, ok := annotations[annotationName]; ok {
  939. annotationNames = append(annotationNames, fmt.Sprintf("%s=%s", annotationName, val))
  940. } else if indexOf(UnallocatedSuffix, annotationNames) == -1 { // if UnallocatedSuffix not already in names
  941. annotationNames = append(annotationNames, UnallocatedSuffix)
  942. }
  943. }
  944. // resolve arbitrary ordering. e.g., app=app0/env=env0 is the same agg as env=env0/app=app0
  945. if len(annotationNames) > 1 {
  946. sort.Strings(annotationNames)
  947. }
  948. unallocatedSuffixIndex := indexOf(UnallocatedSuffix, annotationNames)
  949. // suffix should be at index 0 if it exists b/c of underscores
  950. if unallocatedSuffixIndex != -1 {
  951. annotationNames = append(annotationNames[:unallocatedSuffixIndex], annotationNames[unallocatedSuffixIndex+1:]...)
  952. annotationNames = append(annotationNames, UnallocatedSuffix) // append to end
  953. }
  954. names = append(names, annotationNames...)
  955. }
  956. }
  957. if properties.HasLabel() {
  958. labels, err := alloc.Properties.GetLabels() // labels that the individual allocation possesses
  959. if err != nil {
  960. // Indicate that allocation has no labels
  961. names = append(names, UnallocatedSuffix)
  962. } else {
  963. labelNames := []string{}
  964. aggLabels, err := properties.GetLabels() // potential labels to aggregate on supplied by the API caller
  965. if err != nil {
  966. // We've already checked HasLabel, so this should never occur
  967. return "", err
  968. }
  969. // calvin - support multi-label aggregation
  970. for labelName := range aggLabels {
  971. if val, ok := labels[labelName]; ok {
  972. labelNames = append(labelNames, fmt.Sprintf("%s=%s", labelName, val))
  973. } else if indexOf(UnallocatedSuffix, labelNames) == -1 { // if UnallocatedSuffix not already in names
  974. labelNames = append(labelNames, UnallocatedSuffix)
  975. }
  976. }
  977. // resolve arbitrary ordering. e.g., app=app0/env=env0 is the same agg as env=env0/app=app0
  978. if len(labelNames) > 1 {
  979. sort.Strings(labelNames)
  980. }
  981. unallocatedSuffixIndex := indexOf(UnallocatedSuffix, labelNames)
  982. // suffix should be at index 0 if it exists b/c of underscores
  983. if unallocatedSuffixIndex != -1 {
  984. labelNames = append(labelNames[:unallocatedSuffixIndex], labelNames[unallocatedSuffixIndex+1:]...)
  985. labelNames = append(labelNames, UnallocatedSuffix) // append to end
  986. }
  987. names = append(names, labelNames...)
  988. }
  989. }
  990. return strings.Join(names, "/"), nil
  991. }
  992. // TODO clean up
  993. // Helper function to check for slice membership. Not sure if repeated elsewhere in our codebase.
  994. func indexOf(v string, arr []string) int {
  995. for i, s := range arr {
  996. // This is caseless equivalence
  997. if strings.EqualFold(v, s) {
  998. return i
  999. }
  1000. }
  1001. return -1
  1002. }
  1003. // Clone returns a new AllocationSet with a deep copy of the given
  1004. // AllocationSet's allocations.
  1005. func (as *AllocationSet) Clone() *AllocationSet {
  1006. if as == nil {
  1007. return nil
  1008. }
  1009. as.RLock()
  1010. defer as.RUnlock()
  1011. allocs := map[string]*Allocation{}
  1012. for k, v := range as.allocations {
  1013. allocs[k] = v.Clone()
  1014. }
  1015. externalKeys := map[string]bool{}
  1016. for k, v := range as.externalKeys {
  1017. externalKeys[k] = v
  1018. }
  1019. idleKeys := map[string]bool{}
  1020. for k, v := range as.idleKeys {
  1021. idleKeys[k] = v
  1022. }
  1023. return &AllocationSet{
  1024. allocations: allocs,
  1025. externalKeys: externalKeys,
  1026. idleKeys: idleKeys,
  1027. Window: as.Window.Clone(),
  1028. }
  1029. }
  1030. // ComputeIdleAllocations computes the idle allocations for the AllocationSet,
  1031. // given a set of Assets. Ideally, assetSet should contain only Nodes, but if
  1032. // it contains other Assets, they will be ignored; only CPU, GPU and RAM are
  1033. // considered for idle allocation. One idle allocation per-cluster will be
  1034. // computed and returned, keyed by cluster_id.
  1035. func (as *AllocationSet) ComputeIdleAllocations(assetSet *AssetSet) (map[string]*Allocation, error) {
  1036. if as == nil {
  1037. return nil, fmt.Errorf("cannot compute idle allocation for nil AllocationSet")
  1038. }
  1039. // TODO: external allocation: remove after testing and benchmarking
  1040. profStart := time.Now()
  1041. defer log.Profile(profStart, fmt.Sprintf("ComputeIdleAllocations: %s", as.Window))
  1042. if assetSet == nil {
  1043. return nil, fmt.Errorf("cannot compute idle allocation with nil AssetSet")
  1044. }
  1045. if !as.Window.Equal(assetSet.Window) {
  1046. return nil, fmt.Errorf("cannot compute idle allocation for sets with mismatched windows: %s != %s", as.Window, assetSet.Window)
  1047. }
  1048. window := as.Window
  1049. // Build a map of cumulative cluster asset costs, per resource; i.e.
  1050. // cluster-to-{cpu|gpu|ram}-to-cost.
  1051. assetClusterResourceCosts := map[string]map[string]float64{}
  1052. assetSet.Each(func(key string, a Asset) {
  1053. if node, ok := a.(*Node); ok {
  1054. if _, ok := assetClusterResourceCosts[node.Properties().Cluster]; !ok {
  1055. assetClusterResourceCosts[node.Properties().Cluster] = map[string]float64{}
  1056. }
  1057. assetClusterResourceCosts[node.Properties().Cluster]["cpu"] += node.CPUCost * (1.0 - node.Discount)
  1058. assetClusterResourceCosts[node.Properties().Cluster]["gpu"] += node.GPUCost * (1.0 - node.Discount)
  1059. assetClusterResourceCosts[node.Properties().Cluster]["ram"] += node.RAMCost * (1.0 - node.Discount)
  1060. }
  1061. })
  1062. // Determine start, end on a per-cluster basis
  1063. clusterStarts := map[string]time.Time{}
  1064. clusterEnds := map[string]time.Time{}
  1065. // Subtract allocated costs from asset costs, leaving only the remaining
  1066. // idle costs.
  1067. as.Each(func(name string, a *Allocation) {
  1068. cluster, err := a.Properties.GetCluster()
  1069. if err != nil {
  1070. // Failed to find allocation's cluster
  1071. return
  1072. }
  1073. if _, ok := assetClusterResourceCosts[cluster]; !ok {
  1074. // Failed to find assets for allocation's cluster
  1075. return
  1076. }
  1077. // Set cluster (start, end) if they are either not currently set,
  1078. // or if the detected (start, end) of the current allocation falls
  1079. // before or after, respectively, the current values.
  1080. if s, ok := clusterStarts[cluster]; !ok || a.Start.Before(s) {
  1081. clusterStarts[cluster] = a.Start
  1082. }
  1083. if e, ok := clusterEnds[cluster]; !ok || a.End.Before(e) {
  1084. clusterEnds[cluster] = a.End
  1085. }
  1086. assetClusterResourceCosts[cluster]["cpu"] -= a.CPUCost
  1087. assetClusterResourceCosts[cluster]["gpu"] -= a.GPUCost
  1088. assetClusterResourceCosts[cluster]["ram"] -= a.RAMCost
  1089. })
  1090. // Turn remaining un-allocated asset costs into idle allocations
  1091. idleAllocs := map[string]*Allocation{}
  1092. for cluster, resources := range assetClusterResourceCosts {
  1093. // Default start and end to the (start, end) of the given window, but
  1094. // use the actual, detected (start, end) pair if they are available.
  1095. start := *window.Start()
  1096. if s, ok := clusterStarts[cluster]; ok && window.Contains(s) {
  1097. start = s
  1098. }
  1099. end := *window.End()
  1100. if e, ok := clusterEnds[cluster]; ok && window.Contains(e) {
  1101. end = e
  1102. }
  1103. idleAlloc := &Allocation{
  1104. Name: fmt.Sprintf("%s/%s", cluster, IdleSuffix),
  1105. Window: window.Clone(),
  1106. Properties: Properties{ClusterProp: cluster},
  1107. Start: start,
  1108. End: end,
  1109. CPUCost: resources["cpu"],
  1110. GPUCost: resources["gpu"],
  1111. RAMCost: resources["ram"],
  1112. }
  1113. idleAlloc.TotalCost = idleAlloc.CPUCost + idleAlloc.GPUCost + idleAlloc.RAMCost
  1114. // Do not continue if multiple idle allocations are computed for a
  1115. // single cluster.
  1116. if _, ok := idleAllocs[cluster]; ok {
  1117. return nil, fmt.Errorf("duplicate idle allocations for cluster %s", cluster)
  1118. }
  1119. idleAllocs[cluster] = idleAlloc
  1120. }
  1121. return idleAllocs, nil
  1122. }
  1123. // Delete removes the allocation with the given name from the set
  1124. func (as *AllocationSet) Delete(name string) {
  1125. if as == nil {
  1126. return
  1127. }
  1128. as.Lock()
  1129. defer as.Unlock()
  1130. delete(as.externalKeys, name)
  1131. delete(as.idleKeys, name)
  1132. delete(as.allocations, name)
  1133. }
  1134. // Each invokes the given function for each Allocation in the set
  1135. func (as *AllocationSet) Each(f func(string, *Allocation)) {
  1136. if as == nil {
  1137. return
  1138. }
  1139. for k, a := range as.allocations {
  1140. f(k, a)
  1141. }
  1142. }
  1143. // End returns the End time of the AllocationSet window
  1144. func (as *AllocationSet) End() time.Time {
  1145. if as == nil {
  1146. log.Warningf("Allocation ETL: calling End on nil AllocationSet")
  1147. return time.Unix(0, 0)
  1148. }
  1149. if as.Window.End() == nil {
  1150. log.Warningf("Allocation ETL: AllocationSet with illegal window: End is nil; len(as.allocations)=%d", len(as.allocations))
  1151. return time.Unix(0, 0)
  1152. }
  1153. return *as.Window.End()
  1154. }
  1155. // Get returns the Allocation at the given key in the AllocationSet
  1156. func (as *AllocationSet) Get(key string) *Allocation {
  1157. as.RLock()
  1158. defer as.RUnlock()
  1159. if alloc, ok := as.allocations[key]; ok {
  1160. return alloc
  1161. }
  1162. return nil
  1163. }
  1164. // ExternalAllocations returns a map of the external allocations in the set.
  1165. // Returns clones of the actual Allocations, so mutability is not a problem.
  1166. func (as *AllocationSet) ExternalAllocations() map[string]*Allocation {
  1167. externals := map[string]*Allocation{}
  1168. if as.IsEmpty() {
  1169. return externals
  1170. }
  1171. as.RLock()
  1172. defer as.RUnlock()
  1173. for key := range as.externalKeys {
  1174. if alloc, ok := as.allocations[key]; ok {
  1175. externals[key] = alloc.Clone()
  1176. }
  1177. }
  1178. return externals
  1179. }
  1180. // ExternalCost returns the total aggregated external costs of the set
  1181. func (as *AllocationSet) ExternalCost() float64 {
  1182. if as.IsEmpty() {
  1183. return 0.0
  1184. }
  1185. as.RLock()
  1186. defer as.RUnlock()
  1187. externalCost := 0.0
  1188. for _, alloc := range as.allocations {
  1189. externalCost += alloc.ExternalCost
  1190. }
  1191. return externalCost
  1192. }
  1193. // IdleAllocations returns a map of the idle allocations in the AllocationSet.
  1194. // Returns clones of the actual Allocations, so mutability is not a problem.
  1195. func (as *AllocationSet) IdleAllocations() map[string]*Allocation {
  1196. idles := map[string]*Allocation{}
  1197. if as.IsEmpty() {
  1198. return idles
  1199. }
  1200. as.RLock()
  1201. defer as.RUnlock()
  1202. for key := range as.idleKeys {
  1203. if alloc, ok := as.allocations[key]; ok {
  1204. idles[key] = alloc.Clone()
  1205. }
  1206. }
  1207. return idles
  1208. }
  1209. // Insert aggregates the current entry in the AllocationSet by the given Allocation,
  1210. // but only if the Allocation is valid, i.e. matches the AllocationSet's window. If
  1211. // there is no existing entry, one is created. Nil error response indicates success.
  1212. func (as *AllocationSet) Insert(that *Allocation) error {
  1213. return as.insert(that)
  1214. }
  1215. func (as *AllocationSet) insert(that *Allocation) error {
  1216. if as == nil {
  1217. return fmt.Errorf("cannot insert into nil AllocationSet")
  1218. }
  1219. as.Lock()
  1220. defer as.Unlock()
  1221. if as.allocations == nil {
  1222. as.allocations = map[string]*Allocation{}
  1223. }
  1224. if as.externalKeys == nil {
  1225. as.externalKeys = map[string]bool{}
  1226. }
  1227. if as.idleKeys == nil {
  1228. as.idleKeys = map[string]bool{}
  1229. }
  1230. // Add the given Allocation to the existing entry, if there is one;
  1231. // otherwise just set directly into allocations
  1232. if _, ok := as.allocations[that.Name]; !ok {
  1233. as.allocations[that.Name] = that
  1234. } else {
  1235. as.allocations[that.Name].add(that)
  1236. }
  1237. // If the given Allocation is an external one, record that
  1238. if that.IsExternal() {
  1239. as.externalKeys[that.Name] = true
  1240. }
  1241. // If the given Allocation is an idle one, record that
  1242. if that.IsIdle() {
  1243. as.idleKeys[that.Name] = true
  1244. }
  1245. return nil
  1246. }
  1247. // IsEmpty returns true if the AllocationSet is nil, or if it contains
  1248. // zero allocations.
  1249. func (as *AllocationSet) IsEmpty() bool {
  1250. if as == nil || len(as.allocations) == 0 {
  1251. return true
  1252. }
  1253. as.RLock()
  1254. defer as.RUnlock()
  1255. return as.allocations == nil || len(as.allocations) == 0
  1256. }
  1257. // Length returns the number of Allocations in the set
  1258. func (as *AllocationSet) Length() int {
  1259. if as == nil {
  1260. return 0
  1261. }
  1262. as.RLock()
  1263. defer as.RUnlock()
  1264. return len(as.allocations)
  1265. }
  1266. // Map clones and returns a map of the AllocationSet's Allocations
  1267. func (as *AllocationSet) Map() map[string]*Allocation {
  1268. if as.IsEmpty() {
  1269. return map[string]*Allocation{}
  1270. }
  1271. return as.Clone().allocations
  1272. }
  1273. // MarshalJSON JSON-encodes the AllocationSet
  1274. func (as *AllocationSet) MarshalJSON() ([]byte, error) {
  1275. as.RLock()
  1276. defer as.RUnlock()
  1277. return json.Marshal(as.allocations)
  1278. }
  1279. // Resolution returns the AllocationSet's window duration
  1280. func (as *AllocationSet) Resolution() time.Duration {
  1281. return as.Window.Duration()
  1282. }
  1283. // Set uses the given Allocation to overwrite the existing entry in the
  1284. // AllocationSet under the Allocation's name.
  1285. func (as *AllocationSet) Set(alloc *Allocation) error {
  1286. if as.IsEmpty() {
  1287. as.Lock()
  1288. as.allocations = map[string]*Allocation{}
  1289. as.externalKeys = map[string]bool{}
  1290. as.idleKeys = map[string]bool{}
  1291. as.Unlock()
  1292. }
  1293. as.Lock()
  1294. defer as.Unlock()
  1295. as.allocations[alloc.Name] = alloc
  1296. // If the given Allocation is an external one, record that
  1297. if alloc.IsExternal() {
  1298. as.externalKeys[alloc.Name] = true
  1299. }
  1300. // If the given Allocation is an idle one, record that
  1301. if alloc.IsIdle() {
  1302. as.idleKeys[alloc.Name] = true
  1303. }
  1304. return nil
  1305. }
  1306. // Start returns the Start time of the AllocationSet window
  1307. func (as *AllocationSet) Start() time.Time {
  1308. if as == nil {
  1309. log.Warningf("Allocation ETL: calling Start on nil AllocationSet")
  1310. return time.Unix(0, 0)
  1311. }
  1312. if as.Window.Start() == nil {
  1313. log.Warningf("Allocation ETL: AllocationSet with illegal window: Start is nil; len(as.allocations)=%d", len(as.allocations))
  1314. return time.Unix(0, 0)
  1315. }
  1316. return *as.Window.Start()
  1317. }
  1318. // String represents the given Allocation as a string
  1319. func (as *AllocationSet) String() string {
  1320. if as == nil {
  1321. return "<nil>"
  1322. }
  1323. return fmt.Sprintf("AllocationSet{length: %d; window: %s; totalCost: %.2f}",
  1324. as.Length(), as.Window, as.TotalCost())
  1325. }
  1326. // TotalCost returns the sum of all TotalCosts of the allocations contained
  1327. func (as *AllocationSet) TotalCost() float64 {
  1328. if as.IsEmpty() {
  1329. return 0.0
  1330. }
  1331. as.RLock()
  1332. defer as.RUnlock()
  1333. tc := 0.0
  1334. for _, a := range as.allocations {
  1335. tc += a.TotalCost
  1336. }
  1337. return tc
  1338. }
  1339. // UTCOffset returns the AllocationSet's configured UTCOffset.
  1340. func (as *AllocationSet) UTCOffset() time.Duration {
  1341. _, zone := as.Start().Zone()
  1342. return time.Duration(zone) * time.Second
  1343. }
  1344. func (as *AllocationSet) accumulate(that *AllocationSet) (*AllocationSet, error) {
  1345. if as.IsEmpty() {
  1346. return that, nil
  1347. }
  1348. if that.IsEmpty() {
  1349. return as, nil
  1350. }
  1351. // TODO niko/cdmr implement first
  1352. // if that.Window.Overlaps(as.Window) {
  1353. // return nil, fmt.Errorf("AllocationSet.accumulate: overlapping windows: %s", that.Window, as.Window)
  1354. // }
  1355. // Set start, end to min(start), max(end)
  1356. start := as.Start()
  1357. end := as.End()
  1358. if that.Start().Before(start) {
  1359. start = that.Start()
  1360. }
  1361. if that.End().After(end) {
  1362. end = that.End()
  1363. }
  1364. acc := NewAllocationSet(start, end)
  1365. as.RLock()
  1366. defer as.RUnlock()
  1367. that.RLock()
  1368. defer that.RUnlock()
  1369. for _, alloc := range as.allocations {
  1370. err := acc.insert(alloc)
  1371. if err != nil {
  1372. return nil, err
  1373. }
  1374. }
  1375. for _, alloc := range that.allocations {
  1376. err := acc.insert(alloc)
  1377. if err != nil {
  1378. return nil, err
  1379. }
  1380. }
  1381. return acc, nil
  1382. }
  1383. // AllocationSetRange is a thread-safe slice of AllocationSets. It is meant to
  1384. // be used such that the AllocationSets held are consecutive and coherent with
  1385. // respect to using the same aggregation properties, UTC offset, and
  1386. // resolution. However these rules are not necessarily enforced, so use wisely.
  1387. type AllocationSetRange struct {
  1388. sync.RWMutex
  1389. allocations []*AllocationSet
  1390. }
  1391. // NewAllocationSetRange instantiates a new range composed of the given
  1392. // AllocationSets in the order provided.
  1393. func NewAllocationSetRange(allocs ...*AllocationSet) *AllocationSetRange {
  1394. return &AllocationSetRange{
  1395. allocations: allocs,
  1396. }
  1397. }
  1398. // Accumulate sums each AllocationSet in the given range, returning a single cumulative
  1399. // AllocationSet for the entire range.
  1400. func (asr *AllocationSetRange) Accumulate() (*AllocationSet, error) {
  1401. var allocSet *AllocationSet
  1402. var err error
  1403. asr.RLock()
  1404. defer asr.RUnlock()
  1405. for _, as := range asr.allocations {
  1406. allocSet, err = allocSet.accumulate(as)
  1407. if err != nil {
  1408. return nil, err
  1409. }
  1410. }
  1411. return allocSet, nil
  1412. }
  1413. // TODO niko/etl accumulate into lower-resolution chunks of the given resolution
  1414. // func (asr *AllocationSetRange) AccumulateBy(resolution time.Duration) *AllocationSetRange
  1415. // AggregateBy aggregates each AllocationSet in the range by the given
  1416. // properties and options.
  1417. func (asr *AllocationSetRange) AggregateBy(properties Properties, options *AllocationAggregationOptions) error {
  1418. aggRange := &AllocationSetRange{allocations: []*AllocationSet{}}
  1419. asr.Lock()
  1420. defer asr.Unlock()
  1421. for _, as := range asr.allocations {
  1422. err := as.AggregateBy(properties, options)
  1423. if err != nil {
  1424. return err
  1425. }
  1426. aggRange.allocations = append(aggRange.allocations, as)
  1427. }
  1428. asr.allocations = aggRange.allocations
  1429. return nil
  1430. }
  1431. // Append appends the given AllocationSet to the end of the range. It does not
  1432. // validate whether or not that violates window continuity.
  1433. func (asr *AllocationSetRange) Append(that *AllocationSet) {
  1434. asr.Lock()
  1435. defer asr.Unlock()
  1436. asr.allocations = append(asr.allocations, that)
  1437. }
  1438. // Each invokes the given function for each AllocationSet in the range
  1439. func (asr *AllocationSetRange) Each(f func(int, *AllocationSet)) {
  1440. if asr == nil {
  1441. return
  1442. }
  1443. for i, as := range asr.allocations {
  1444. f(i, as)
  1445. }
  1446. }
  1447. // Get retrieves the AllocationSet at the given index of the range.
  1448. func (asr *AllocationSetRange) Get(i int) (*AllocationSet, error) {
  1449. if i < 0 || i >= len(asr.allocations) {
  1450. return nil, fmt.Errorf("AllocationSetRange: index out of range: %d", i)
  1451. }
  1452. asr.RLock()
  1453. defer asr.RUnlock()
  1454. return asr.allocations[i], nil
  1455. }
  1456. // InsertRange merges the given AllocationSetRange into the receiving one by
  1457. // lining up sets with matching windows, then inserting each allocation from
  1458. // the given ASR into the respective set in the receiving ASR. If the given
  1459. // ASR contains an AllocationSet from a window that does not exist in the
  1460. // receiving ASR, then an error is returned. However, the given ASR does not
  1461. // need to cover the full range of the receiver.
  1462. func (asr *AllocationSetRange) InsertRange(that *AllocationSetRange) error {
  1463. if asr == nil {
  1464. return fmt.Errorf("cannot insert range into nil AllocationSetRange")
  1465. }
  1466. // keys maps window to index in asr
  1467. keys := map[string]int{}
  1468. asr.Each(func(i int, as *AllocationSet) {
  1469. if as == nil {
  1470. return
  1471. }
  1472. keys[as.Window.String()] = i
  1473. })
  1474. // Nothing to merge, so simply return
  1475. if len(keys) == 0 {
  1476. return nil
  1477. }
  1478. var err error
  1479. that.Each(func(j int, thatAS *AllocationSet) {
  1480. if thatAS == nil || err != nil {
  1481. return
  1482. }
  1483. // Find matching AllocationSet in asr
  1484. i, ok := keys[thatAS.Window.String()]
  1485. if !ok {
  1486. err = fmt.Errorf("cannot merge AllocationSet into window that does not exist: %s", thatAS.Window.String())
  1487. return
  1488. }
  1489. as, err := asr.Get(i)
  1490. if err != nil {
  1491. err = fmt.Errorf("AllocationSetRange index does not exist: %d", i)
  1492. return
  1493. }
  1494. // Insert each Allocation from the given set
  1495. thatAS.Each(func(k string, alloc *Allocation) {
  1496. err = as.Insert(alloc)
  1497. if err != nil {
  1498. err = fmt.Errorf("error inserting allocation: %s", err)
  1499. return
  1500. }
  1501. })
  1502. })
  1503. // err might be nil
  1504. return err
  1505. }
  1506. // Length returns the length of the range, which is zero if nil
  1507. func (asr *AllocationSetRange) Length() int {
  1508. if asr == nil || asr.allocations == nil {
  1509. return 0
  1510. }
  1511. asr.RLock()
  1512. defer asr.RUnlock()
  1513. return len(asr.allocations)
  1514. }
  1515. // MarshalJSON JSON-encodes the range
  1516. func (asr *AllocationSetRange) MarshalJSON() ([]byte, error) {
  1517. asr.RLock()
  1518. asr.RUnlock()
  1519. return json.Marshal(asr.allocations)
  1520. }
  1521. // Slice copies the underlying slice of AllocationSets, maintaining order,
  1522. // and returns the copied slice.
  1523. func (asr *AllocationSetRange) Slice() []*AllocationSet {
  1524. if asr == nil || asr.allocations == nil {
  1525. return nil
  1526. }
  1527. asr.RLock()
  1528. defer asr.RUnlock()
  1529. copy := []*AllocationSet{}
  1530. for _, as := range asr.allocations {
  1531. copy = append(copy, as.Clone())
  1532. }
  1533. return copy
  1534. }
  1535. // String represents the given AllocationSetRange as a string
  1536. func (asr *AllocationSetRange) String() string {
  1537. if asr == nil {
  1538. return "<nil>"
  1539. }
  1540. return fmt.Sprintf("AllocationSetRange{length: %d}", asr.Length())
  1541. }
  1542. // UTCOffset returns the detected UTCOffset of the AllocationSets within the
  1543. // range. Defaults to 0 if the range is nil or empty. Does not warn if there
  1544. // are sets with conflicting UTCOffsets (just returns the first).
  1545. func (asr *AllocationSetRange) UTCOffset() time.Duration {
  1546. if asr.Length() == 0 {
  1547. return 0
  1548. }
  1549. as, err := asr.Get(0)
  1550. if err != nil {
  1551. return 0
  1552. }
  1553. return as.UTCOffset()
  1554. }
  1555. // Window returns the full window that the AllocationSetRange spans, from the
  1556. // start of the first AllocationSet to the end of the last one.
  1557. func (asr *AllocationSetRange) Window() Window {
  1558. if asr == nil || asr.Length() == 0 {
  1559. return NewWindow(nil, nil)
  1560. }
  1561. start := asr.allocations[0].Start()
  1562. end := asr.allocations[asr.Length()-1].End()
  1563. return NewWindow(&start, &end)
  1564. }