allocation.go 48 KB

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