allocation.go 52 KB

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