summaryallocation.go 54 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773
  1. package kubecost
  2. import (
  3. "errors"
  4. "fmt"
  5. "strings"
  6. "sync"
  7. "time"
  8. "github.com/opencost/opencost/pkg/log"
  9. "github.com/opencost/opencost/pkg/util/timeutil"
  10. )
  11. // SummaryAllocation summarizes an Allocation, keeping only fields necessary
  12. // for providing a high-level view of identifying the Allocation over a period
  13. // of time (Start, End) over which it ran, and inspecting the associated per-
  14. // resource costs (subtotaled with adjustments), total cost, and efficiency.
  15. //
  16. // SummaryAllocation does not have a concept of Window (i.e. the time period
  17. // within which it is defined, as opposed to the Start and End times). That
  18. // context must be provided by a SummaryAllocationSet.
  19. type SummaryAllocation struct {
  20. Name string `json:"name"`
  21. Properties *AllocationProperties `json:"-"`
  22. Start time.Time `json:"start"`
  23. End time.Time `json:"end"`
  24. CPUCoreRequestAverage float64 `json:"cpuCoreRequestAverage"`
  25. CPUCoreUsageAverage float64 `json:"cpuCoreUsageAverage"`
  26. CPUCost float64 `json:"cpuCost"`
  27. GPUCost float64 `json:"gpuCost"`
  28. NetworkCost float64 `json:"networkCost"`
  29. LoadBalancerCost float64 `json:"loadBalancerCost"`
  30. PVCost float64 `json:"pvCost"`
  31. RAMBytesRequestAverage float64 `json:"ramByteRequestAverage"`
  32. RAMBytesUsageAverage float64 `json:"ramByteUsageAverage"`
  33. RAMCost float64 `json:"ramCost"`
  34. SharedCost float64 `json:"sharedCost"`
  35. ExternalCost float64 `json:"externalCost"`
  36. Share bool `json:"-"`
  37. }
  38. // NewSummaryAllocation converts an Allocation to a SummaryAllocation by
  39. // dropping unnecessary fields and consolidating others (e.g. adjustments).
  40. // Reconciliation happens here because that process is synonymous with the
  41. // consolidation of adjustment fields.
  42. func NewSummaryAllocation(alloc *Allocation, reconcile, reconcileNetwork bool) *SummaryAllocation {
  43. if alloc == nil {
  44. return nil
  45. }
  46. sa := &SummaryAllocation{
  47. Name: alloc.Name,
  48. Properties: alloc.Properties,
  49. Start: alloc.Start,
  50. End: alloc.End,
  51. CPUCoreRequestAverage: alloc.CPUCoreRequestAverage,
  52. CPUCoreUsageAverage: alloc.CPUCoreUsageAverage,
  53. CPUCost: alloc.CPUCost + alloc.CPUCostAdjustment,
  54. GPUCost: alloc.GPUCost + alloc.GPUCostAdjustment,
  55. NetworkCost: alloc.NetworkCost + alloc.NetworkCostAdjustment,
  56. LoadBalancerCost: alloc.LoadBalancerCost + alloc.LoadBalancerCostAdjustment,
  57. PVCost: alloc.PVCost() + alloc.PVCostAdjustment,
  58. RAMBytesRequestAverage: alloc.RAMBytesRequestAverage,
  59. RAMBytesUsageAverage: alloc.RAMBytesUsageAverage,
  60. RAMCost: alloc.RAMCost + alloc.RAMCostAdjustment,
  61. SharedCost: alloc.SharedCost,
  62. ExternalCost: alloc.ExternalCost,
  63. }
  64. // Revert adjustments if reconciliation is off. If only network
  65. // reconciliation is off, only revert network adjustment.
  66. if !reconcile {
  67. sa.CPUCost -= alloc.CPUCostAdjustment
  68. sa.GPUCost -= alloc.GPUCostAdjustment
  69. sa.NetworkCost -= alloc.NetworkCostAdjustment
  70. sa.LoadBalancerCost -= alloc.LoadBalancerCostAdjustment
  71. sa.PVCost -= alloc.PVCostAdjustment
  72. sa.RAMCost -= alloc.RAMCostAdjustment
  73. } else if !reconcileNetwork {
  74. sa.NetworkCost -= alloc.NetworkCostAdjustment
  75. }
  76. return sa
  77. }
  78. // Add sums two SummaryAllocations, adding the given SummaryAllocation to the
  79. // receiving one, thus mutating the receiver. For performance reasons, it
  80. // simply drops Properties, so a SummaryAllocation can only be Added once.
  81. func (sa *SummaryAllocation) Add(that *SummaryAllocation) error {
  82. if sa == nil || that == nil {
  83. return errors.New("cannot Add a nil SummaryAllocation")
  84. }
  85. // Once Added, a SummaryAllocation has no Properties. This saves us from
  86. // having to compute the intersection of two sets of Properties, which is
  87. // expensive.
  88. sa.Properties = nil
  89. // Sum non-cumulative fields by turning them into cumulative, adding them,
  90. // and then converting them back into averages after minutes have been
  91. // combined (just below).
  92. cpuReqCoreMins := sa.CPUCoreRequestAverage * sa.Minutes()
  93. cpuReqCoreMins += that.CPUCoreRequestAverage * that.Minutes()
  94. cpuUseCoreMins := sa.CPUCoreUsageAverage * sa.Minutes()
  95. cpuUseCoreMins += that.CPUCoreUsageAverage * that.Minutes()
  96. ramReqByteMins := sa.RAMBytesRequestAverage * sa.Minutes()
  97. ramReqByteMins += that.RAMBytesRequestAverage * that.Minutes()
  98. ramUseByteMins := sa.RAMBytesUsageAverage * sa.Minutes()
  99. ramUseByteMins += that.RAMBytesUsageAverage * that.Minutes()
  100. // Expand Start and End to be the "max" of among the given Allocations
  101. if that.Start.Before(sa.Start) {
  102. sa.Start = that.Start
  103. }
  104. if that.End.After(sa.End) {
  105. sa.End = that.End
  106. }
  107. // Convert cumulative request and usage back into rates
  108. if sa.Minutes() > 0 {
  109. sa.CPUCoreRequestAverage = cpuReqCoreMins / sa.Minutes()
  110. sa.CPUCoreUsageAverage = cpuUseCoreMins / sa.Minutes()
  111. sa.RAMBytesRequestAverage = ramReqByteMins / sa.Minutes()
  112. sa.RAMBytesUsageAverage = ramUseByteMins / sa.Minutes()
  113. } else {
  114. sa.CPUCoreRequestAverage = 0.0
  115. sa.CPUCoreUsageAverage = 0.0
  116. sa.RAMBytesRequestAverage = 0.0
  117. sa.RAMBytesUsageAverage = 0.0
  118. }
  119. // Sum all cumulative cost fields
  120. sa.CPUCost += that.CPUCost
  121. sa.ExternalCost += that.ExternalCost
  122. sa.GPUCost += that.GPUCost
  123. sa.LoadBalancerCost += that.LoadBalancerCost
  124. sa.NetworkCost += that.NetworkCost
  125. sa.PVCost += that.PVCost
  126. sa.RAMCost += that.RAMCost
  127. sa.SharedCost += that.SharedCost
  128. return nil
  129. }
  130. // Clone copies the SummaryAllocation and returns the copy
  131. func (sa *SummaryAllocation) Clone() *SummaryAllocation {
  132. return &SummaryAllocation{
  133. Name: sa.Name,
  134. Properties: sa.Properties.Clone(),
  135. Start: sa.Start,
  136. End: sa.End,
  137. CPUCoreRequestAverage: sa.CPUCoreRequestAverage,
  138. CPUCoreUsageAverage: sa.CPUCoreUsageAverage,
  139. CPUCost: sa.CPUCost,
  140. GPUCost: sa.GPUCost,
  141. NetworkCost: sa.NetworkCost,
  142. LoadBalancerCost: sa.LoadBalancerCost,
  143. PVCost: sa.PVCost,
  144. RAMBytesRequestAverage: sa.RAMBytesRequestAverage,
  145. RAMBytesUsageAverage: sa.RAMBytesUsageAverage,
  146. RAMCost: sa.RAMCost,
  147. SharedCost: sa.SharedCost,
  148. ExternalCost: sa.ExternalCost,
  149. }
  150. }
  151. // CPUEfficiency is the ratio of usage to request. If there is no request and
  152. // no usage or cost, then efficiency is zero. If there is no request, but there
  153. // is usage or cost, then efficiency is 100%.
  154. func (sa *SummaryAllocation) CPUEfficiency() float64 {
  155. if sa == nil || sa.IsIdle() {
  156. return 0.0
  157. }
  158. if sa.CPUCoreRequestAverage > 0 {
  159. return sa.CPUCoreUsageAverage / sa.CPUCoreRequestAverage
  160. }
  161. if sa.CPUCoreUsageAverage == 0.0 || sa.CPUCost == 0.0 {
  162. return 0.0
  163. }
  164. return 1.0
  165. }
  166. func (sa *SummaryAllocation) Equal(that *SummaryAllocation) bool {
  167. if sa == nil || that == nil {
  168. return false
  169. }
  170. if sa.Name != that.Name {
  171. return false
  172. }
  173. if sa.Start != that.Start {
  174. return false
  175. }
  176. if sa.End != that.End {
  177. return false
  178. }
  179. if sa.CPUCoreRequestAverage != that.CPUCoreRequestAverage {
  180. return false
  181. }
  182. if sa.CPUCoreUsageAverage != that.CPUCoreUsageAverage {
  183. return false
  184. }
  185. if sa.CPUCost != that.CPUCost {
  186. return false
  187. }
  188. if sa.GPUCost != that.GPUCost {
  189. return false
  190. }
  191. if sa.NetworkCost != that.NetworkCost {
  192. return false
  193. }
  194. if sa.LoadBalancerCost != that.LoadBalancerCost {
  195. return false
  196. }
  197. if sa.PVCost != that.PVCost {
  198. return false
  199. }
  200. if sa.RAMBytesRequestAverage != that.RAMBytesRequestAverage {
  201. return false
  202. }
  203. if sa.RAMBytesUsageAverage != that.RAMBytesUsageAverage {
  204. return false
  205. }
  206. if sa.RAMCost != that.RAMCost {
  207. return false
  208. }
  209. if sa.SharedCost != that.SharedCost {
  210. return false
  211. }
  212. if sa.ExternalCost != that.ExternalCost {
  213. return false
  214. }
  215. return true
  216. }
  217. func (sa *SummaryAllocation) generateKey(aggregateBy []string, labelConfig *LabelConfig) string {
  218. if sa == nil {
  219. return ""
  220. }
  221. return sa.Properties.GenerateKey(aggregateBy, labelConfig)
  222. }
  223. // IsExternal is true if the given SummaryAllocation represents external costs.
  224. func (sa *SummaryAllocation) IsExternal() bool {
  225. if sa == nil {
  226. return false
  227. }
  228. return strings.Contains(sa.Name, ExternalSuffix)
  229. }
  230. // IsIdle is true if the given SummaryAllocation represents idle costs.
  231. func (sa *SummaryAllocation) IsIdle() bool {
  232. if sa == nil {
  233. return false
  234. }
  235. return strings.Contains(sa.Name, IdleSuffix)
  236. }
  237. // IsUnallocated is true if the given SummaryAllocation represents unallocated
  238. // costs.
  239. func (sa *SummaryAllocation) IsUnallocated() bool {
  240. if sa == nil {
  241. return false
  242. }
  243. return strings.Contains(sa.Name, UnallocatedSuffix)
  244. }
  245. // IsUnmounted is true if the given SummaryAllocation represents unmounted
  246. // volume costs.
  247. func (sa *SummaryAllocation) IsUnmounted() bool {
  248. if sa == nil {
  249. return false
  250. }
  251. return strings.Contains(sa.Name, UnmountedSuffix)
  252. }
  253. // Minutes returns the number of minutes the SummaryAllocation represents, as
  254. // defined by the difference between the end and start times.
  255. func (sa *SummaryAllocation) Minutes() float64 {
  256. if sa == nil {
  257. return 0.0
  258. }
  259. return sa.End.Sub(sa.Start).Minutes()
  260. }
  261. // RAMEfficiency is the ratio of usage to request. If there is no request and
  262. // no usage or cost, then efficiency is zero. If there is no request, but there
  263. // is usage or cost, then efficiency is 100%.
  264. func (sa *SummaryAllocation) RAMEfficiency() float64 {
  265. if sa == nil || sa.IsIdle() {
  266. return 0.0
  267. }
  268. if sa.RAMBytesRequestAverage > 0 {
  269. return sa.RAMBytesUsageAverage / sa.RAMBytesRequestAverage
  270. }
  271. if sa.RAMBytesUsageAverage == 0.0 || sa.RAMCost == 0.0 {
  272. return 0.0
  273. }
  274. return 1.0
  275. }
  276. // TotalCost is the total cost of the SummaryAllocation
  277. func (sa *SummaryAllocation) TotalCost() float64 {
  278. if sa == nil {
  279. return 0.0
  280. }
  281. return sa.CPUCost + sa.GPUCost + sa.RAMCost + sa.PVCost + sa.NetworkCost + sa.LoadBalancerCost + sa.SharedCost + sa.ExternalCost
  282. }
  283. // TotalEfficiency is the cost-weighted average of CPU and RAM efficiency. If
  284. // there is no cost at all, then efficiency is zero.
  285. func (sa *SummaryAllocation) TotalEfficiency() float64 {
  286. if sa == nil || sa.IsIdle() {
  287. return 0.0
  288. }
  289. if sa.RAMCost+sa.CPUCost > 0 {
  290. ramCostEff := sa.RAMEfficiency() * sa.RAMCost
  291. cpuCostEff := sa.CPUEfficiency() * sa.CPUCost
  292. return (ramCostEff + cpuCostEff) / (sa.CPUCost + sa.RAMCost)
  293. }
  294. return 0.0
  295. }
  296. // SummaryAllocationSet stores a set of SummaryAllocations, each with a unique
  297. // name, that share a window. An AllocationSet is mutable, so treat it like a
  298. // threadsafe map.
  299. type SummaryAllocationSet struct {
  300. sync.RWMutex
  301. externalKeys map[string]bool
  302. idleKeys map[string]bool
  303. SummaryAllocations map[string]*SummaryAllocation `json:"allocations"`
  304. Window Window `json:"window"`
  305. }
  306. // NewSummaryAllocationSet converts an AllocationSet to a SummaryAllocationSet.
  307. // Filter functions, keep functions, and reconciliation parameters are
  308. // required for unfortunate reasons to do with performance and legacy order-of-
  309. // operations details, as well as the fact that reconciliation has been
  310. // pushed down to the conversion step between Allocation and SummaryAllocation.
  311. func NewSummaryAllocationSet(as *AllocationSet, filter AllocationFilter, kfs []AllocationMatchFunc, reconcile, reconcileNetwork bool) *SummaryAllocationSet {
  312. if as == nil {
  313. return nil
  314. }
  315. // Pre-flatten the filter so we can just check == nil to see if there are
  316. // filters.
  317. if filter != nil {
  318. filter = filter.Flattened()
  319. }
  320. // If we can know the exact size of the map, use it. If filters or sharing
  321. // functions are present, we can't know the size, so we make a default map.
  322. var sasMap map[string]*SummaryAllocation
  323. if filter == nil && len(kfs) == 0 {
  324. // No filters, so make the map of summary allocations exactly the size
  325. // of the origin allocation set.
  326. sasMap = make(map[string]*SummaryAllocation, len(as.Allocations))
  327. } else {
  328. // There are filters, so start with a standard map
  329. sasMap = make(map[string]*SummaryAllocation)
  330. }
  331. sas := &SummaryAllocationSet{
  332. SummaryAllocations: sasMap,
  333. Window: as.Window.Clone(),
  334. }
  335. for _, alloc := range as.Allocations {
  336. // First, detect if the allocation should be kept. If so, mark it as
  337. // such, insert it, and continue.
  338. shouldKeep := false
  339. for _, kf := range kfs {
  340. if kf(alloc) {
  341. shouldKeep = true
  342. break
  343. }
  344. }
  345. if shouldKeep {
  346. sa := NewSummaryAllocation(alloc, reconcile, reconcileNetwork)
  347. sa.Share = true
  348. sas.Insert(sa)
  349. continue
  350. }
  351. // If the allocation does not pass any of the given filter functions,
  352. // do not insert it into the set.
  353. if filter != nil && !filter.Matches(alloc) {
  354. continue
  355. }
  356. err := sas.Insert(NewSummaryAllocation(alloc, reconcile, reconcileNetwork))
  357. if err != nil {
  358. log.Errorf("SummaryAllocation: error inserting summary of %s", alloc.Name)
  359. }
  360. }
  361. for key := range as.ExternalKeys {
  362. sas.externalKeys[key] = true
  363. }
  364. for key := range as.IdleKeys {
  365. sas.idleKeys[key] = true
  366. }
  367. return sas
  368. }
  369. // Clone creates a deep copy of the SummaryAllocationSet
  370. func (sas *SummaryAllocationSet) Clone() *SummaryAllocationSet {
  371. sas.RLock()
  372. defer sas.RUnlock()
  373. externalKeys := make(map[string]bool, len(sas.externalKeys))
  374. for k, v := range sas.externalKeys {
  375. externalKeys[k] = v
  376. }
  377. idleKeys := make(map[string]bool, len(sas.idleKeys))
  378. for k, v := range sas.idleKeys {
  379. idleKeys[k] = v
  380. }
  381. summaryAllocations := make(map[string]*SummaryAllocation, len(sas.SummaryAllocations))
  382. for k, v := range sas.SummaryAllocations {
  383. summaryAllocations[k] = v.Clone()
  384. }
  385. return &SummaryAllocationSet{
  386. externalKeys: externalKeys,
  387. idleKeys: idleKeys,
  388. SummaryAllocations: summaryAllocations,
  389. Window: sas.Window.Clone(),
  390. }
  391. }
  392. // Add sums two SummaryAllocationSets, which Adds all SummaryAllocations in the
  393. // given SummaryAllocationSet to thier counterparts in the receiving set. Add
  394. // also expands the Window to include both constituent Windows, in the case
  395. // that Add is being used from accumulating (as opposed to aggregating). For
  396. // performance reasons, the function may return either a new set, or an
  397. // unmodified original, so it should not be assumed that the original sets are
  398. // safeuly usable after calling Add.
  399. func (sas *SummaryAllocationSet) Add(that *SummaryAllocationSet) (*SummaryAllocationSet, error) {
  400. if sas == nil || len(sas.SummaryAllocations) == 0 {
  401. return that, nil
  402. }
  403. if that == nil || len(that.SummaryAllocations) == 0 {
  404. return sas, nil
  405. }
  406. if sas.Window.IsOpen() {
  407. return nil, errors.New("cannot add a SummaryAllocationSet with an open window")
  408. }
  409. // Set start, end to min(start), max(end)
  410. start := *sas.Window.Start()
  411. end := *sas.Window.End()
  412. if that.Window.Start().Before(start) {
  413. start = *that.Window.Start()
  414. }
  415. if that.Window.End().After(end) {
  416. end = *that.Window.End()
  417. }
  418. acc := &SummaryAllocationSet{
  419. SummaryAllocations: make(map[string]*SummaryAllocation, len(sas.SummaryAllocations)),
  420. Window: NewClosedWindow(start, end),
  421. }
  422. sas.RLock()
  423. defer sas.RUnlock()
  424. that.RLock()
  425. defer that.RUnlock()
  426. for _, alloc := range sas.SummaryAllocations {
  427. err := acc.Insert(alloc)
  428. if err != nil {
  429. return nil, err
  430. }
  431. }
  432. for _, alloc := range that.SummaryAllocations {
  433. err := acc.Insert(alloc)
  434. if err != nil {
  435. return nil, err
  436. }
  437. }
  438. return acc, nil
  439. }
  440. // AggregateBy aggregates the Allocations in the given AllocationSet by the given
  441. // AllocationProperty. This will only be legal if the AllocationSet is divisible by the
  442. // given AllocationProperty; e.g. Containers can be divided by Namespace, but not vice-a-versa.
  443. func (sas *SummaryAllocationSet) AggregateBy(aggregateBy []string, options *AllocationAggregationOptions) error {
  444. if sas == nil || len(sas.SummaryAllocations) == 0 {
  445. return nil
  446. }
  447. if sas.Window.IsOpen() {
  448. return errors.New("cannot aggregate a SummaryAllocationSet with an open window")
  449. }
  450. if options == nil {
  451. options = &AllocationAggregationOptions{}
  452. }
  453. if options.LabelConfig == nil {
  454. options.LabelConfig = NewLabelConfig()
  455. }
  456. // Pre-flatten the filter so we can just check == nil to see if there are
  457. // filters.
  458. if options.Filter != nil {
  459. options.Filter = options.Filter.Flattened()
  460. }
  461. // Check if we have any work to do; if not, then early return. If
  462. // aggregateBy is nil, we don't aggregate anything. On the other hand,
  463. // an empty slice implies that we should aggregate everything. (See
  464. // generateKey for why that makes sense.)
  465. shouldAggregate := aggregateBy != nil
  466. shouldKeep := len(options.SharedHourlyCosts) > 0 || len(options.ShareFuncs) > 0
  467. if !shouldAggregate && !shouldKeep {
  468. return nil
  469. }
  470. // The order of operations for aggregating a SummaryAllotionSet is as
  471. // follows:
  472. //
  473. // 1. Partition external, idle, and shared allocations into separate sets.
  474. // Also, create the resultSet into which the results will be aggregated.
  475. //
  476. // 2. Record resource totals for shared costs and unmounted volumes so
  477. // that we can account for them in computing idle coefficients.
  478. //
  479. // 3. Retrieve pre-computed allocation resource totals, which will be used
  480. // to compute idle sharing coefficients.
  481. //
  482. // 4. Convert shared hourly cost into a cumulative allocation to share,
  483. // and insert it into the share set.
  484. //
  485. // 5. Compute sharing coefficients per-aggregation, if sharing resources.
  486. //
  487. // 6. Distribute idle allocations according to the idle coefficients.
  488. //
  489. // 7. Record allocation resource totals (after filtration) if filters have
  490. // been applied. (Used for filtering proportional amount of idle.)
  491. //
  492. // 8. Generate aggregation key and insert allocation into the output set
  493. //
  494. // 9. If idle is shared and resources are shared, it's probable that some
  495. // amount of idle cost will be shared with a shared resource.
  496. // Distribute that idle cost, if it exists, among the respective shared
  497. // allocations before sharing them with the aggregated allocations.
  498. //
  499. // 10. Apply idle filtration, which "filters" the idle cost, or scales it
  500. // by the proportion of allocation resources remaining after filters
  501. // have been applied.
  502. //
  503. // 11. Distribute shared resources according to sharing coefficients.
  504. //
  505. // 12. Insert external allocations into the result set.
  506. //
  507. // 13. Insert any undistributed idle, in the case that idle
  508. // coefficients end up being zero and some idle is not shared.
  509. //
  510. // 14. Combine all idle allocations into a single idle allocation, unless
  511. // the option to keep idle split by cluster or node is enabled.
  512. // 1. Partition external, idle, and shared allocations into separate sets.
  513. // Also, create the resultSet into which the results will be aggregated.
  514. // resultSet will collect the aggregated allocations
  515. resultSet := &SummaryAllocationSet{
  516. Window: sas.Window.Clone(),
  517. }
  518. // externalSet will collect external allocations
  519. externalSet := &SummaryAllocationSet{
  520. Window: sas.Window.Clone(),
  521. }
  522. // idleSet will be shared among resultSet after initial aggregation
  523. // is complete
  524. idleSet := &SummaryAllocationSet{
  525. Window: sas.Window.Clone(),
  526. }
  527. // shareSet will be shared among resultSet after initial aggregation
  528. // is complete
  529. shareSet := &SummaryAllocationSet{
  530. Window: sas.Window.Clone(),
  531. }
  532. sas.Lock()
  533. defer sas.Unlock()
  534. // 2. Record resource totals for shared costs, aggregating by cluster or by
  535. // node (depending on if idle is partitioned by cluster or node) so that we
  536. // can account for them in computing idle coefficients. Do the same for
  537. // unmounted volume costs, which only require a total cost.
  538. sharedResourceTotals := map[string]*AllocationTotals{}
  539. totalUnmountedCost := 0.0
  540. // 1 & 2. Identify set membership and aggregate aforementioned totals.
  541. for _, sa := range sas.SummaryAllocations {
  542. if sa.Share {
  543. var key string
  544. if options.IdleByNode {
  545. key = fmt.Sprintf("%s/%s", sa.Properties.Cluster, sa.Properties.Node)
  546. } else {
  547. key = sa.Properties.Cluster
  548. }
  549. if _, ok := sharedResourceTotals[key]; !ok {
  550. sharedResourceTotals[key] = &AllocationTotals{}
  551. }
  552. sharedResourceTotals[key].CPUCost += sa.CPUCost
  553. sharedResourceTotals[key].GPUCost += sa.GPUCost
  554. sharedResourceTotals[key].LoadBalancerCost += sa.LoadBalancerCost
  555. sharedResourceTotals[key].NetworkCost += sa.NetworkCost
  556. sharedResourceTotals[key].PersistentVolumeCost += sa.PVCost
  557. sharedResourceTotals[key].RAMCost += sa.RAMCost
  558. shareSet.Insert(sa)
  559. delete(sas.SummaryAllocations, sa.Name)
  560. continue
  561. }
  562. // External allocations get aggregated post-hoc (see step 6) and do
  563. // not necessarily contain complete sets of properties, so they are
  564. // moved to a separate AllocationSet.
  565. if sa.IsExternal() {
  566. delete(sas.externalKeys, sa.Name)
  567. delete(sas.SummaryAllocations, sa.Name)
  568. externalSet.Insert(sa)
  569. continue
  570. }
  571. // Idle allocations should be separated into idleSet if they are to be
  572. // shared later on. If they are not to be shared, then add them to the
  573. // resultSet like any other allocation.
  574. if sa.IsIdle() {
  575. delete(sas.idleKeys, sa.Name)
  576. delete(sas.SummaryAllocations, sa.Name)
  577. if options.ShareIdle == ShareEven || options.ShareIdle == ShareWeighted {
  578. idleSet.Insert(sa)
  579. } else {
  580. resultSet.Insert(sa)
  581. }
  582. continue
  583. }
  584. // Track total unmounted cost because it must be taken out of total
  585. // allocated costs for sharing coefficients.
  586. if sa.IsUnmounted() {
  587. totalUnmountedCost += sa.TotalCost()
  588. }
  589. }
  590. // It's possible that no more un-shared, non-idle, non-external allocations
  591. // remain at this point. This always results in an emptySet, so return early.
  592. if len(sas.SummaryAllocations) == 0 {
  593. sas.SummaryAllocations = map[string]*SummaryAllocation{}
  594. return nil
  595. }
  596. // 3. Retrieve pre-computed allocation resource totals, which will be used
  597. // to compute idle coefficients, based on the ratio of an allocation's per-
  598. // resource cost to the per-resource totals of that allocation's cluster or
  599. // node. Whether to perform this operation based on cluster or node is an
  600. // option. (See IdleByNode documentation; defaults to idle-by-cluster.)
  601. var allocTotals map[string]*AllocationTotals
  602. var ok bool
  603. if options.AllocationTotalsStore != nil {
  604. if options.IdleByNode {
  605. allocTotals, ok = options.AllocationTotalsStore.GetAllocationTotalsByNode(*sas.Window.Start(), *sas.Window.End())
  606. if !ok {
  607. return fmt.Errorf("nil allocation resource totals by node for %s", sas.Window)
  608. }
  609. } else {
  610. allocTotals, ok = options.AllocationTotalsStore.GetAllocationTotalsByCluster(*sas.Window.Start(), *sas.Window.End())
  611. if !ok {
  612. return fmt.Errorf("nil allocation resource totals by cluster for %s", sas.Window)
  613. }
  614. }
  615. }
  616. // If reconciliation has been fully or partially disabled, clear the
  617. // relevant adjustments from the alloc totals
  618. if allocTotals != nil && (!options.Reconcile || !options.ReconcileNetwork) {
  619. if !options.Reconcile {
  620. for _, tot := range allocTotals {
  621. tot.ClearAdjustments()
  622. }
  623. } else if !options.ReconcileNetwork {
  624. for _, tot := range allocTotals {
  625. tot.NetworkCostAdjustment = 0.0
  626. }
  627. }
  628. }
  629. // If filters have been applied, then we need to record allocation resource
  630. // totals after filtration (i.e. the allocations that are present) so that
  631. // we can identify the proportion of idle cost to keep. That is, we should
  632. // only return the idle cost that would be shared with the remaining
  633. // allocations, even if we're keeping idle separate. The totals should be
  634. // recorded by idle-key (cluster or node, depending on the IdleByNode
  635. // option). Instantiating this map is a signal to record the totals.
  636. var allocTotalsAfterFilters map[string]*AllocationTotals
  637. if len(resultSet.idleKeys) > 0 && options.Filter != nil {
  638. allocTotalsAfterFilters = make(map[string]*AllocationTotals, len(resultSet.idleKeys))
  639. }
  640. // If we're recording allocTotalsAfterFilters and there are shared costs,
  641. // then record those resource totals here so that idle for those shared
  642. // resources gets included.
  643. if allocTotalsAfterFilters != nil {
  644. for key, rt := range sharedResourceTotals {
  645. if _, ok := allocTotalsAfterFilters[key]; !ok {
  646. allocTotalsAfterFilters[key] = &AllocationTotals{}
  647. }
  648. // Record only those fields required for computing idle
  649. allocTotalsAfterFilters[key].CPUCost += rt.CPUCost
  650. allocTotalsAfterFilters[key].GPUCost += rt.GPUCost
  651. allocTotalsAfterFilters[key].RAMCost += rt.RAMCost
  652. }
  653. }
  654. // 4. Convert shared hourly cost into a cumulative allocation to share,
  655. // and insert it into the share set.
  656. for name, cost := range options.SharedHourlyCosts {
  657. if cost > 0.0 {
  658. hours := sas.Window.Hours()
  659. // If set ends in the future, adjust hours accordingly
  660. diff := time.Since(*sas.Window.End())
  661. if diff < 0.0 {
  662. hours += diff.Hours()
  663. }
  664. totalSharedCost := cost * hours
  665. shareSet.Insert(&SummaryAllocation{
  666. Name: fmt.Sprintf("%s/%s", name, SharedSuffix),
  667. Properties: &AllocationProperties{},
  668. Start: *sas.Window.Start(),
  669. End: *sas.Window.End(),
  670. SharedCost: totalSharedCost,
  671. })
  672. }
  673. }
  674. // Sharing coefficients are recorded by post-aggregation-key (e.g. if
  675. // aggregating by namespace, then the key will be the namespace) and only
  676. // need to be recorded if there are shared resources. Instantiating this
  677. // map is the signal to record sharing coefficients.
  678. var sharingCoeffs map[string]float64
  679. if len(shareSet.SummaryAllocations) > 0 {
  680. sharingCoeffs = map[string]float64{}
  681. }
  682. // Loop over all remaining SummaryAllocations (after filters, sharing, &c.)
  683. // doing the following, in this order:
  684. // 5. Compute sharing coefficients, if there are shared resources
  685. // 6. Distribute idle cost, if sharing idle
  686. // 7. Record allocTotalsAfterFiltration, if filters have been applied
  687. // 8. Aggregate by key
  688. for _, sa := range sas.SummaryAllocations {
  689. // Generate key to use for aggregation-by-key and allocation name
  690. key := sa.generateKey(aggregateBy, options.LabelConfig)
  691. // 5. Incrementally add to sharing coefficients before adding idle
  692. // cost, which would skew the coefficients. These coefficients will be
  693. // later divided by a total, turning them into a coefficient between
  694. // 0.0 and 1.0.
  695. // NOTE: SummaryAllocation does not support ShareEven, so only record
  696. // by cost for cost-weighted distribution.
  697. if sharingCoeffs != nil {
  698. sharingCoeffs[key] += sa.TotalCost() - sa.SharedCost
  699. }
  700. // 6. Distribute idle allocations according to the idle coefficients.
  701. // NOTE: if idle allocation is off (i.e. options.ShareIdle: ShareNone)
  702. // then all idle allocations will be in the resultSet at this point, so
  703. // idleSet will be empty and we won't enter this block.
  704. if len(idleSet.SummaryAllocations) > 0 {
  705. for _, idle := range idleSet.SummaryAllocations {
  706. // Idle key is either cluster or node, as determined by the
  707. // IdleByNode option.
  708. var key string
  709. // Only share idle allocation with current allocation (sa) if
  710. // the relevant properties match (i.e. cluster and/or node)
  711. if idle.Properties.Cluster != sa.Properties.Cluster {
  712. continue
  713. }
  714. key = idle.Properties.Cluster
  715. if options.IdleByNode {
  716. if idle.Properties.Node != sa.Properties.Node {
  717. continue
  718. }
  719. key = fmt.Sprintf("%s/%s", idle.Properties.Cluster, idle.Properties.Node)
  720. }
  721. cpuCoeff, gpuCoeff, ramCoeff := ComputeIdleCoefficients(options.ShareIdle, key, sa.CPUCost, sa.GPUCost, sa.RAMCost, allocTotals)
  722. sa.CPUCost += idle.CPUCost * cpuCoeff
  723. sa.GPUCost += idle.GPUCost * gpuCoeff
  724. sa.RAMCost += idle.RAMCost * ramCoeff
  725. }
  726. }
  727. // The key becomes the allocation's name, which is used as the key by
  728. // which the allocation is inserted into the set.
  729. sa.Name = key
  730. // If merging unallocated allocations, rename all unallocated
  731. // allocations as simply __unallocated__
  732. if options.MergeUnallocated && sa.IsUnallocated() {
  733. sa.Name = UnallocatedSuffix
  734. }
  735. // 7. Record filtered resource totals for idle allocation filtration,
  736. // only if necessary.
  737. if allocTotalsAfterFilters != nil {
  738. key := sa.Properties.Cluster
  739. if options.IdleByNode {
  740. key = fmt.Sprintf("%s/%s", sa.Properties.Cluster, sa.Properties.Node)
  741. }
  742. if _, ok := allocTotalsAfterFilters[key]; !ok {
  743. allocTotalsAfterFilters[key] = &AllocationTotals{}
  744. }
  745. allocTotalsAfterFilters[key].CPUCost += sa.CPUCost
  746. allocTotalsAfterFilters[key].GPUCost += sa.GPUCost
  747. allocTotalsAfterFilters[key].RAMCost += sa.RAMCost
  748. }
  749. // 8. Inserting the allocation with the generated key for a name
  750. // performs the actual aggregation step.
  751. resultSet.Insert(sa)
  752. }
  753. // 9. If idle is shared and resources are shared, it's probable that some
  754. // amount of idle cost will be shared with a shared resource. Distribute
  755. // that idle cost, if it exists, among the respective shared allocations
  756. // before sharing them with the aggregated allocations.
  757. if len(idleSet.SummaryAllocations) > 0 && len(shareSet.SummaryAllocations) > 0 {
  758. for _, sa := range shareSet.SummaryAllocations {
  759. for _, idle := range idleSet.SummaryAllocations {
  760. var key string
  761. // Only share idle allocation with current allocation (sa) if
  762. // the relevant property matches (i.e. Cluster or Node,
  763. // depending on which idle sharing option is selected)
  764. if options.IdleByNode {
  765. if idle.Properties.Cluster != sa.Properties.Cluster || idle.Properties.Node != sa.Properties.Node {
  766. continue
  767. }
  768. key = fmt.Sprintf("%s/%s", idle.Properties.Cluster, idle.Properties.Node)
  769. } else {
  770. if idle.Properties.Cluster != sa.Properties.Cluster {
  771. continue
  772. }
  773. key = idle.Properties.Cluster
  774. }
  775. cpuCoeff, gpuCoeff, ramCoeff := ComputeIdleCoefficients(options.ShareIdle, key, sa.CPUCost, sa.GPUCost, sa.RAMCost, allocTotals)
  776. sa.CPUCost += idle.CPUCost * cpuCoeff
  777. sa.GPUCost += idle.GPUCost * gpuCoeff
  778. sa.RAMCost += idle.RAMCost * ramCoeff
  779. }
  780. }
  781. }
  782. // 10. Apply idle filtration, which "filters" the idle cost, i.e. scales
  783. // idle allocation costs per-resource by the proportion of allocation
  784. // resources remaining after filtering. In effect, this returns only the
  785. // idle costs that would have been shared with the remaining allocations,
  786. // even if idle is kept separated.
  787. if allocTotalsAfterFilters != nil {
  788. for idleKey := range resultSet.idleKeys {
  789. ia := resultSet.SummaryAllocations[idleKey]
  790. var key string
  791. if options.IdleByNode {
  792. key = fmt.Sprintf("%s/%s", ia.Properties.Cluster, ia.Properties.Node)
  793. } else {
  794. key = ia.Properties.Cluster
  795. }
  796. // Percentage of idle that should remain after filters are applied,
  797. // which equals the proportion of filtered-to-actual cost.
  798. cpuFilterCoeff := 0.0
  799. if allocTotals[key].TotalCPUCost() > 0.0 {
  800. filteredAlloc, ok := allocTotalsAfterFilters[key]
  801. if ok {
  802. cpuFilterCoeff = filteredAlloc.TotalCPUCost() / allocTotals[key].TotalCPUCost()
  803. } else {
  804. cpuFilterCoeff = 0.0
  805. }
  806. }
  807. gpuFilterCoeff := 0.0
  808. if allocTotals[key].TotalGPUCost() > 0.0 {
  809. filteredAlloc, ok := allocTotalsAfterFilters[key]
  810. if ok {
  811. gpuFilterCoeff = filteredAlloc.TotalGPUCost() / allocTotals[key].TotalGPUCost()
  812. } else {
  813. gpuFilterCoeff = 0.0
  814. }
  815. }
  816. ramFilterCoeff := 0.0
  817. if allocTotals[key].TotalRAMCost() > 0.0 {
  818. filteredAlloc, ok := allocTotalsAfterFilters[key]
  819. if ok {
  820. ramFilterCoeff = filteredAlloc.TotalRAMCost() / allocTotals[key].TotalRAMCost()
  821. } else {
  822. ramFilterCoeff = 0.0
  823. }
  824. }
  825. ia.CPUCost *= cpuFilterCoeff
  826. ia.GPUCost *= gpuFilterCoeff
  827. ia.RAMCost *= ramFilterCoeff
  828. }
  829. }
  830. // 11. Distribute shared resources according to sharing coefficients.
  831. // NOTE: ShareEven is not supported
  832. if len(shareSet.SummaryAllocations) > 0 {
  833. sharingCoeffDenominator := 0.0
  834. for _, rt := range allocTotals {
  835. sharingCoeffDenominator += rt.TotalCost()
  836. }
  837. // Do not include the shared costs, themselves, when determining
  838. // sharing coefficients.
  839. for _, rt := range sharedResourceTotals {
  840. sharingCoeffDenominator -= rt.TotalCost()
  841. }
  842. // Do not include the unmounted costs when determining sharing
  843. // coefficients becuase they do not receive shared costs.
  844. sharingCoeffDenominator -= totalUnmountedCost
  845. if sharingCoeffDenominator <= 0.0 {
  846. log.Warnf("SummaryAllocation: sharing coefficient denominator is %f", sharingCoeffDenominator)
  847. } else {
  848. // Compute sharing coeffs by dividing the thus-far accumulated
  849. // numerators by the now-finalized denominator.
  850. for key := range sharingCoeffs {
  851. if sharingCoeffs[key] > 0.0 {
  852. sharingCoeffs[key] /= sharingCoeffDenominator
  853. } else {
  854. log.Warnf("SummaryAllocation: detected illegal sharing coefficient for %s: %v (setting to zero)", key, sharingCoeffs[key])
  855. sharingCoeffs[key] = 0.0
  856. }
  857. }
  858. for key, sa := range resultSet.SummaryAllocations {
  859. // Idle and unmounted allocations, by definition, do not
  860. // receive shared cost
  861. if sa.IsIdle() || sa.IsUnmounted() {
  862. continue
  863. }
  864. sharingCoeff := sharingCoeffs[key]
  865. // Distribute each shared cost with the current allocation on the
  866. // basis of the proportion of the allocation's cost (ShareWeighted)
  867. // or count (ShareEven) to the total aggregated cost or count. This
  868. // condition should hold in spite of filters because the sharing
  869. // coefficient denominator is held constant by pre-computed
  870. // resource totals and the post-aggregation total cost of the
  871. // remaining allocations will, by definition, not be affected.
  872. for _, shared := range shareSet.SummaryAllocations {
  873. sa.SharedCost += shared.TotalCost() * sharingCoeff
  874. }
  875. }
  876. }
  877. }
  878. // 12. Insert external allocations into the result set.
  879. for _, sa := range externalSet.SummaryAllocations {
  880. skip := false
  881. // Make an allocation with the same properties and test that
  882. // against the FilterFunc to see if the external allocation should
  883. // be filtered or not.
  884. // TODO:CLEANUP do something about external cost, this stinks
  885. ea := &Allocation{Properties: sa.Properties}
  886. if options.Filter != nil {
  887. skip = !options.Filter.Matches(ea)
  888. }
  889. if !skip {
  890. key := sa.generateKey(aggregateBy, options.LabelConfig)
  891. sa.Name = key
  892. resultSet.Insert(sa)
  893. }
  894. }
  895. // 13. Distribute remaining, undistributed idle. Undistributed idle is any
  896. // per-resource idle cost for which there can be no idle coefficient
  897. // computed because there is zero usage across all allocations.
  898. for _, isa := range idleSet.SummaryAllocations {
  899. // if the idle does not apply to the non-filtered values, skip it
  900. skip := false
  901. // Make an allocation with the same properties and test that
  902. // against the FilterFunc to see if the external allocation should
  903. // be filtered or not.
  904. // TODO:CLEANUP do something about external cost, this stinks
  905. ia := &Allocation{Properties: isa.Properties}
  906. if options.Filter != nil {
  907. skip = !options.Filter.Matches(ia)
  908. }
  909. if skip {
  910. continue
  911. }
  912. key := isa.Properties.Cluster
  913. if options.IdleByNode {
  914. key = fmt.Sprintf("%s/%s", isa.Properties.Cluster, isa.Properties.Node)
  915. }
  916. rt, ok := allocTotals[key]
  917. if !ok {
  918. log.Warnf("SummaryAllocation: AggregateBy: cannot handle undistributed idle for '%s'", key)
  919. continue
  920. }
  921. hasUndistributableCost := false
  922. if isa.CPUCost > 0.0 && rt.CPUCost == 0.0 {
  923. // There is idle CPU cost, but no allocated CPU cost, so that cost
  924. // is undistributable and must be inserted.
  925. hasUndistributableCost = true
  926. } else {
  927. // Cost was entirely distributed, so zero it out
  928. isa.CPUCost = 0.0
  929. }
  930. if isa.GPUCost > 0.0 && rt.GPUCost == 0.0 {
  931. // There is idle GPU cost, but no allocated GPU cost, so that cost
  932. // is undistributable and must be inserted.
  933. hasUndistributableCost = true
  934. } else {
  935. // Cost was entirely distributed, so zero it out
  936. isa.GPUCost = 0.0
  937. }
  938. if isa.RAMCost > 0.0 && rt.RAMCost == 0.0 {
  939. // There is idle CPU cost, but no allocated CPU cost, so that cost
  940. // is undistributable and must be inserted.
  941. hasUndistributableCost = true
  942. } else {
  943. // Cost was entirely distributed, so zero it out
  944. isa.RAMCost = 0.0
  945. }
  946. if hasUndistributableCost {
  947. isa.Name = fmt.Sprintf("%s/%s", key, IdleSuffix)
  948. resultSet.Insert(isa)
  949. }
  950. }
  951. // 14. Combine all idle allocations into a single idle allocation, unless
  952. // the option to keep idle split by cluster or node is enabled.
  953. if !options.SplitIdle {
  954. for _, ia := range resultSet.idleAllocations() {
  955. resultSet.Delete(ia.Name)
  956. ia.Name = IdleSuffix
  957. resultSet.Insert(ia)
  958. }
  959. }
  960. // Replace the existing set's data with the new, aggregated summary data
  961. sas.SummaryAllocations = resultSet.SummaryAllocations
  962. return nil
  963. }
  964. // Delete removes the allocation with the given name from the set
  965. func (sas *SummaryAllocationSet) Delete(name string) {
  966. if sas == nil {
  967. return
  968. }
  969. sas.Lock()
  970. defer sas.Unlock()
  971. delete(sas.externalKeys, name)
  972. delete(sas.idleKeys, name)
  973. delete(sas.SummaryAllocations, name)
  974. }
  975. // Each invokes the given function for each SummaryAllocation in the set
  976. func (sas *SummaryAllocationSet) Each(f func(string, *SummaryAllocation)) {
  977. if sas == nil {
  978. return
  979. }
  980. for k, a := range sas.SummaryAllocations {
  981. f(k, a)
  982. }
  983. }
  984. func (sas *SummaryAllocationSet) Equal(that *SummaryAllocationSet) bool {
  985. if sas == nil || that == nil {
  986. return false
  987. }
  988. sas.RLock()
  989. defer sas.RUnlock()
  990. if !sas.Window.Equal(that.Window) {
  991. return false
  992. }
  993. if len(sas.SummaryAllocations) != len(that.SummaryAllocations) {
  994. return false
  995. }
  996. for name, sa := range sas.SummaryAllocations {
  997. thatSA, ok := that.SummaryAllocations[name]
  998. if !ok {
  999. return false
  1000. }
  1001. if !sa.Equal(thatSA) {
  1002. return false
  1003. }
  1004. }
  1005. return true
  1006. }
  1007. // IdleAllocations returns a map of the idle allocations in the AllocationSet.
  1008. func (sas *SummaryAllocationSet) idleAllocations() map[string]*SummaryAllocation {
  1009. idles := map[string]*SummaryAllocation{}
  1010. if sas == nil || len(sas.SummaryAllocations) == 0 {
  1011. return idles
  1012. }
  1013. sas.RLock()
  1014. defer sas.RUnlock()
  1015. for key := range sas.idleKeys {
  1016. if sa, ok := sas.SummaryAllocations[key]; ok {
  1017. idles[key] = sa
  1018. }
  1019. }
  1020. return idles
  1021. }
  1022. // Insert aggregates the current entry in the SummaryAllocationSet by the given Allocation,
  1023. // but only if the Allocation is valid, i.e. matches the SummaryAllocationSet's window. If
  1024. // there is no existing entry, one is created. Nil error response indicates success.
  1025. func (sas *SummaryAllocationSet) Insert(sa *SummaryAllocation) error {
  1026. if sas == nil {
  1027. return fmt.Errorf("cannot insert into nil SummaryAllocationSet")
  1028. }
  1029. if sa == nil {
  1030. return fmt.Errorf("cannot insert a nil SummaryAllocation")
  1031. }
  1032. sas.Lock()
  1033. defer sas.Unlock()
  1034. if sas.SummaryAllocations == nil {
  1035. sas.SummaryAllocations = map[string]*SummaryAllocation{}
  1036. }
  1037. if sas.externalKeys == nil {
  1038. sas.externalKeys = map[string]bool{}
  1039. }
  1040. if sas.idleKeys == nil {
  1041. sas.idleKeys = map[string]bool{}
  1042. }
  1043. // Add the given Allocation to the existing entry, if there is one;
  1044. // otherwise just set directly into allocations
  1045. if _, ok := sas.SummaryAllocations[sa.Name]; ok {
  1046. err := sas.SummaryAllocations[sa.Name].Add(sa)
  1047. if err != nil {
  1048. return fmt.Errorf("SummaryAllocationSet.Insert: error trying to Add: %s", err)
  1049. }
  1050. } else {
  1051. sas.SummaryAllocations[sa.Name] = sa
  1052. }
  1053. // If the given Allocation is an external one, record that
  1054. if sa.IsExternal() {
  1055. sas.externalKeys[sa.Name] = true
  1056. }
  1057. // If the given Allocation is an idle one, record that
  1058. if sa.IsIdle() {
  1059. sas.idleKeys[sa.Name] = true
  1060. }
  1061. return nil
  1062. }
  1063. func (sas *SummaryAllocationSet) TotalCost() float64 {
  1064. if sas == nil {
  1065. return 0.0
  1066. }
  1067. sas.RLock()
  1068. defer sas.RUnlock()
  1069. tc := 0.0
  1070. for _, sa := range sas.SummaryAllocations {
  1071. tc += sa.TotalCost()
  1072. }
  1073. return tc
  1074. }
  1075. // RAMEfficiency func to calculate average RAM efficiency over SummaryAllocationSet
  1076. func (sas *SummaryAllocationSet) RAMEfficiency() float64 {
  1077. if sas == nil {
  1078. return 0.0
  1079. }
  1080. sas.RLock()
  1081. defer sas.RUnlock()
  1082. totalRAMBytesMinutesUsage := 0.0
  1083. totalRAMBytesMinutesRequest := 0.0
  1084. totalRAMCost := 0.0
  1085. for _, sa := range sas.SummaryAllocations {
  1086. if sa.IsIdle() {
  1087. continue
  1088. }
  1089. totalRAMBytesMinutesUsage += sa.RAMBytesUsageAverage * sa.Minutes()
  1090. totalRAMBytesMinutesRequest += sa.RAMBytesRequestAverage * sa.Minutes()
  1091. totalRAMCost += sa.RAMCost
  1092. }
  1093. if totalRAMBytesMinutesRequest > 0 {
  1094. return totalRAMBytesMinutesUsage / totalRAMBytesMinutesRequest
  1095. }
  1096. if totalRAMBytesMinutesUsage == 0.0 || totalRAMCost == 0.0 {
  1097. return 0.0
  1098. }
  1099. return 1.0
  1100. }
  1101. // CPUEfficiency func to calculate average CPU efficiency over SummaryAllocationSet
  1102. func (sas *SummaryAllocationSet) CPUEfficiency() float64 {
  1103. if sas == nil {
  1104. return 0.0
  1105. }
  1106. sas.RLock()
  1107. defer sas.RUnlock()
  1108. totalCPUCoreMinutesUsage := 0.0
  1109. totalCPUCoreMinutesRequest := 0.0
  1110. totalCPUCost := 0.0
  1111. for _, sa := range sas.SummaryAllocations {
  1112. if sa.IsIdle() {
  1113. continue
  1114. }
  1115. totalCPUCoreMinutesUsage += sa.CPUCoreUsageAverage * sa.Minutes()
  1116. totalCPUCoreMinutesRequest += sa.CPUCoreRequestAverage * sa.Minutes()
  1117. totalCPUCost += sa.CPUCost
  1118. }
  1119. if totalCPUCoreMinutesRequest > 0 {
  1120. return totalCPUCoreMinutesUsage / totalCPUCoreMinutesRequest
  1121. }
  1122. if totalCPUCoreMinutesUsage == 0.0 || totalCPUCost == 0.0 {
  1123. return 0.0
  1124. }
  1125. return 1.0
  1126. }
  1127. // TotalEfficiency func to calculate average Total efficiency over SummaryAllocationSet
  1128. func (sas *SummaryAllocationSet) TotalEfficiency() float64 {
  1129. if sas == nil {
  1130. return 0.0
  1131. }
  1132. sas.RLock()
  1133. defer sas.RUnlock()
  1134. totalRAMCost := 0.0
  1135. totalCPUCost := 0.0
  1136. for _, sa := range sas.SummaryAllocations {
  1137. if sa.IsIdle() {
  1138. continue
  1139. }
  1140. totalRAMCost += sa.RAMCost
  1141. totalCPUCost += sa.CPUCost
  1142. }
  1143. if totalRAMCost+totalCPUCost > 0 {
  1144. return (totalRAMCost*sas.RAMEfficiency() + totalCPUCost*sas.CPUEfficiency()) / (totalRAMCost + totalCPUCost)
  1145. }
  1146. return 0.0
  1147. }
  1148. // SummaryAllocationSetRange is a thread-safe slice of SummaryAllocationSets.
  1149. type SummaryAllocationSetRange struct {
  1150. sync.RWMutex
  1151. Step time.Duration `json:"step"`
  1152. SummaryAllocationSets []*SummaryAllocationSet `json:"sets"`
  1153. Window Window `json:"window"`
  1154. Message string `json:"-"`
  1155. }
  1156. // NewSummaryAllocationSetRange instantiates a new range composed of the given
  1157. // SummaryAllocationSets in the order provided. The expectations about the
  1158. // SummaryAllocationSets are as follows:
  1159. // - window durations are all equal
  1160. // - sets are consecutive (i.e. chronologically sorted)
  1161. // - there are no gaps between sets
  1162. // - sets do not have overlapping windows
  1163. func NewSummaryAllocationSetRange(sass ...*SummaryAllocationSet) *SummaryAllocationSetRange {
  1164. var step time.Duration
  1165. window := NewWindow(nil, nil)
  1166. for _, sas := range sass {
  1167. if window.Start() == nil || (sas.Window.Start() != nil && sas.Window.Start().Before(*window.Start())) {
  1168. window.start = sas.Window.Start()
  1169. }
  1170. if window.End() == nil || (sas.Window.End() != nil && sas.Window.End().After(*window.End())) {
  1171. window.end = sas.Window.End()
  1172. }
  1173. if step == 0 {
  1174. step = sas.Window.Duration()
  1175. } else if step != sas.Window.Duration() {
  1176. log.Warnf("instantiating range with step %s using set of step %s is illegal", step, sas.Window.Duration())
  1177. }
  1178. }
  1179. return &SummaryAllocationSetRange{
  1180. Step: step,
  1181. SummaryAllocationSets: sass,
  1182. Window: window,
  1183. }
  1184. }
  1185. // Accumulate sums each AllocationSet in the given range, returning a single cumulative
  1186. // AllocationSet for the entire range.
  1187. func (sasr *SummaryAllocationSetRange) accumulate() (*SummaryAllocationSet, error) {
  1188. var result *SummaryAllocationSet
  1189. var err error
  1190. sasr.RLock()
  1191. defer sasr.RUnlock()
  1192. for _, sas := range sasr.SummaryAllocationSets {
  1193. result, err = result.Add(sas)
  1194. if err != nil {
  1195. return nil, err
  1196. }
  1197. }
  1198. return result, nil
  1199. }
  1200. // newAccumulation clones the first available SummaryAllocationSet to use as the data structure to
  1201. // accumulate the remaining data. This leaves the original SummaryAllocationSetRange intact.
  1202. func (sasr *SummaryAllocationSetRange) newAccumulation() (*SummaryAllocationSet, error) {
  1203. var result *SummaryAllocationSet
  1204. var err error
  1205. sasr.RLock()
  1206. defer sasr.RUnlock()
  1207. for _, sas := range sasr.SummaryAllocationSets {
  1208. // we want to clone the first summary allocation set, then just Add the others
  1209. // to the clone. We will eventually use the clone to create the set range.
  1210. if result == nil {
  1211. result = sas.Clone()
  1212. continue
  1213. }
  1214. // Copy if sas is non-nil
  1215. var sasCopy *SummaryAllocationSet = nil
  1216. if sas != nil {
  1217. sasCopy = sas.Clone()
  1218. }
  1219. // nil is ok to pass into Add
  1220. result, err = result.Add(sasCopy)
  1221. if err != nil {
  1222. return nil, err
  1223. }
  1224. }
  1225. return result, nil
  1226. }
  1227. // AggregateBy aggregates each AllocationSet in the range by the given
  1228. // properties and options.
  1229. func (sasr *SummaryAllocationSetRange) AggregateBy(aggregateBy []string, options *AllocationAggregationOptions) error {
  1230. sasr.Lock()
  1231. defer sasr.Unlock()
  1232. for _, sas := range sasr.SummaryAllocationSets {
  1233. err := sas.AggregateBy(aggregateBy, options)
  1234. if err != nil {
  1235. // Wipe out data so that corrupt data cannot be mistakenly used
  1236. sasr.SummaryAllocationSets = []*SummaryAllocationSet{}
  1237. return err
  1238. }
  1239. }
  1240. return nil
  1241. }
  1242. // Append appends the given AllocationSet to the end of the range. It does not
  1243. // validate whether or not that violates window continuity.
  1244. func (sasr *SummaryAllocationSetRange) Append(sas *SummaryAllocationSet) {
  1245. sasr.Lock()
  1246. defer sasr.Unlock()
  1247. // Append to list of sets
  1248. sasr.SummaryAllocationSets = append(sasr.SummaryAllocationSets, sas)
  1249. // Set step, if not set
  1250. if sasr.Step == 0 {
  1251. sasr.Step = sas.Window.Duration()
  1252. }
  1253. // Adjust window
  1254. if sasr.Window.Start() == nil || (sas.Window.Start() != nil && sas.Window.Start().Before(*sasr.Window.Start())) {
  1255. sasr.Window.start = sas.Window.Start()
  1256. }
  1257. if sasr.Window.End() == nil || (sas.Window.End() != nil && sas.Window.End().After(*sasr.Window.End())) {
  1258. sasr.Window.end = sas.Window.End()
  1259. }
  1260. }
  1261. // Each invokes the given function for each AllocationSet in the range
  1262. func (sasr *SummaryAllocationSetRange) Each(f func(int, *SummaryAllocationSet)) {
  1263. if sasr == nil {
  1264. return
  1265. }
  1266. for i, as := range sasr.SummaryAllocationSets {
  1267. f(i, as)
  1268. }
  1269. }
  1270. // InsertExternalAllocations takes all allocations in the given
  1271. // AllocationSetRange (they should all be considered "external") and inserts
  1272. // them into the receiving SummaryAllocationSetRange.
  1273. // TODO:CLEANUP replace this with a better idea (or get rid of external
  1274. // allocations, as such, altogether)
  1275. func (sasr *SummaryAllocationSetRange) InsertExternalAllocations(that *AllocationSetRange) error {
  1276. if sasr == nil {
  1277. return fmt.Errorf("cannot insert range into nil AllocationSetRange")
  1278. }
  1279. // keys maps window to index in range
  1280. keys := map[string]int{}
  1281. for i, as := range sasr.SummaryAllocationSets {
  1282. if as == nil {
  1283. continue
  1284. }
  1285. keys[as.Window.String()] = i
  1286. }
  1287. // Nothing to merge, so simply return
  1288. if len(keys) == 0 {
  1289. return nil
  1290. }
  1291. var err error
  1292. for _, thatAS := range that.Allocations {
  1293. if thatAS == nil || err != nil {
  1294. continue
  1295. }
  1296. // Find matching AllocationSet in asr
  1297. i, ok := keys[thatAS.Window.String()]
  1298. if !ok {
  1299. err = fmt.Errorf("cannot merge AllocationSet into window that does not exist: %s", thatAS.Window.String())
  1300. continue
  1301. }
  1302. sas := sasr.SummaryAllocationSets[i]
  1303. // Insert each Allocation from the given set
  1304. for _, alloc := range thatAS.Allocations {
  1305. externalSA := NewSummaryAllocation(alloc, true, true)
  1306. // This error will be returned below
  1307. // TODO:CLEANUP should Each have early-error-return functionality?
  1308. err = sas.Insert(externalSA)
  1309. }
  1310. }
  1311. // err might be nil
  1312. return err
  1313. }
  1314. func (sasr *SummaryAllocationSetRange) TotalCost() float64 {
  1315. if sasr == nil {
  1316. return 0.0
  1317. }
  1318. sasr.RLock()
  1319. defer sasr.RUnlock()
  1320. tc := 0.0
  1321. for _, sas := range sasr.SummaryAllocationSets {
  1322. tc += sas.TotalCost()
  1323. }
  1324. return tc
  1325. }
  1326. // TODO remove after testing
  1327. func (sasr *SummaryAllocationSetRange) Print(verbose bool) {
  1328. fmt.Printf("%s (dur=%s, len=%d, cost=%.5f)\n", sasr.Window, sasr.Window.Duration(), len(sasr.SummaryAllocationSets), sasr.TotalCost())
  1329. for _, sas := range sasr.SummaryAllocationSets {
  1330. fmt.Printf(" > %s (dur=%s, len=%d, cost=%.5f) \n", sas.Window, sas.Window.Duration(), len(sas.SummaryAllocations), sas.TotalCost())
  1331. for key, sa := range sas.SummaryAllocations {
  1332. if verbose {
  1333. fmt.Printf(" {\"%s\", cpu: %.5f, gpu: %.5f, lb: %.5f, net: %.5f, pv: %.5f, ram: %.5f, shared: %.5f, external: %.5f}\n",
  1334. key, sa.CPUCost, sa.GPUCost, sa.LoadBalancerCost, sa.NetworkCost, sa.PVCost, sa.RAMCost, sa.SharedCost, sa.ExternalCost)
  1335. } else {
  1336. fmt.Printf(" - \"%s\": %.5f\n", key, sa.TotalCost())
  1337. }
  1338. }
  1339. }
  1340. }
  1341. func (sasr *SummaryAllocationSetRange) Accumulate(accumulateBy AccumulateOption) (*SummaryAllocationSetRange, error) {
  1342. switch accumulateBy {
  1343. case AccumulateOptionNone:
  1344. return sasr.accumulateByNone()
  1345. case AccumulateOptionAll:
  1346. return sasr.accumulateByAll()
  1347. case AccumulateOptionHour:
  1348. return sasr.accumulateByHour()
  1349. case AccumulateOptionDay:
  1350. return sasr.accumulateByDay()
  1351. case AccumulateOptionWeek:
  1352. return sasr.accumulateByWeek()
  1353. case AccumulateOptionMonth:
  1354. return sasr.accumulateByMonth()
  1355. default:
  1356. // this should never happen
  1357. return nil, fmt.Errorf("unexpected error, invalid accumulateByType: %s", accumulateBy)
  1358. }
  1359. }
  1360. func (sasr *SummaryAllocationSetRange) accumulateByNone() (*SummaryAllocationSetRange, error) {
  1361. return sasr.clone(), nil
  1362. }
  1363. func (sasr *SummaryAllocationSetRange) accumulateByAll() (*SummaryAllocationSetRange, error) {
  1364. var err error
  1365. var result *SummaryAllocationSet
  1366. result, err = sasr.newAccumulation()
  1367. if err != nil {
  1368. return nil, fmt.Errorf("error running accumulate: %s", err)
  1369. }
  1370. accumulated := NewSummaryAllocationSetRange(result)
  1371. return accumulated, nil
  1372. }
  1373. func (sasr *SummaryAllocationSetRange) accumulateByHour() (*SummaryAllocationSetRange, error) {
  1374. // ensure that the summary allocation sets have a 1-hour window, if a set exists
  1375. if len(sasr.SummaryAllocationSets) > 0 && sasr.SummaryAllocationSets[0].Window.Duration() != time.Hour {
  1376. return nil, fmt.Errorf("window duration must equal 1 hour; got:%s", sasr.SummaryAllocationSets[0].Window.Duration())
  1377. }
  1378. result := sasr.clone()
  1379. return result, nil
  1380. }
  1381. func (sasr *SummaryAllocationSetRange) accumulateByDay() (*SummaryAllocationSetRange, error) {
  1382. // if the summary allocation set window is 1-day, just return the existing summary allocation set range
  1383. if len(sasr.SummaryAllocationSets) > 0 && sasr.SummaryAllocationSets[0].Window.Duration() == time.Hour*24 {
  1384. return sasr, nil
  1385. }
  1386. var toAccumulate *SummaryAllocationSetRange
  1387. result := NewSummaryAllocationSetRange()
  1388. for i, as := range sasr.SummaryAllocationSets {
  1389. if as.Window.Duration() != time.Hour {
  1390. return nil, fmt.Errorf("window duration must equal 1 hour; got:%s", as.Window.Duration())
  1391. }
  1392. hour := as.Window.Start().Hour()
  1393. if toAccumulate == nil {
  1394. toAccumulate = NewSummaryAllocationSetRange()
  1395. as = as.Clone()
  1396. }
  1397. toAccumulate.Append(as)
  1398. sas, err := toAccumulate.accumulate()
  1399. if err != nil {
  1400. return nil, fmt.Errorf("error accumulating result: %s", err)
  1401. }
  1402. toAccumulate = NewSummaryAllocationSetRange(sas)
  1403. if hour == 23 || i == len(sasr.SummaryAllocationSets)-1 {
  1404. if length := len(toAccumulate.SummaryAllocationSets); length != 1 {
  1405. return nil, fmt.Errorf("failed accumulation, detected %d sets instead of 1", length)
  1406. }
  1407. result.Append(toAccumulate.SummaryAllocationSets[0])
  1408. toAccumulate = nil
  1409. }
  1410. }
  1411. return result, nil
  1412. }
  1413. func (sasr *SummaryAllocationSetRange) accumulateByMonth() (*SummaryAllocationSetRange, error) {
  1414. var toAccumulate *SummaryAllocationSetRange
  1415. result := NewSummaryAllocationSetRange()
  1416. for i, as := range sasr.SummaryAllocationSets {
  1417. if as.Window.Duration() != time.Hour*24 {
  1418. return nil, fmt.Errorf("window duration must equal 24 hours; got:%s", as.Window.Duration())
  1419. }
  1420. _, month, _ := as.Window.Start().Date()
  1421. _, nextDayMonth, _ := as.Window.Start().Add(time.Hour * 24).Date()
  1422. if toAccumulate == nil {
  1423. toAccumulate = NewSummaryAllocationSetRange()
  1424. as = as.Clone()
  1425. }
  1426. toAccumulate.Append(as)
  1427. sas, err := toAccumulate.accumulate()
  1428. if err != nil {
  1429. return nil, fmt.Errorf("error building monthly accumulation: %s", err)
  1430. }
  1431. toAccumulate = NewSummaryAllocationSetRange(sas)
  1432. // either the month has ended, or there are no more summary allocation sets
  1433. if month != nextDayMonth || i == len(sasr.SummaryAllocationSets)-1 {
  1434. if length := len(toAccumulate.SummaryAllocationSets); length != 1 {
  1435. return nil, fmt.Errorf("failed accumulation, detected %d sets instead of 1", length)
  1436. }
  1437. result.Append(toAccumulate.SummaryAllocationSets[0])
  1438. toAccumulate = nil
  1439. }
  1440. }
  1441. return result, nil
  1442. }
  1443. func (sasr *SummaryAllocationSetRange) accumulateByWeek() (*SummaryAllocationSetRange, error) {
  1444. if len(sasr.SummaryAllocationSets) > 0 && sasr.SummaryAllocationSets[0].Window.Duration() == timeutil.Week {
  1445. return sasr, nil
  1446. }
  1447. var toAccumulate *SummaryAllocationSetRange
  1448. result := NewSummaryAllocationSetRange()
  1449. for i, as := range sasr.SummaryAllocationSets {
  1450. if as.Window.Duration() != time.Hour*24 {
  1451. return nil, fmt.Errorf("window duration must equal 24 hours; got:%s", as.Window.Duration())
  1452. }
  1453. dayOfWeek := as.Window.Start().Weekday()
  1454. if toAccumulate == nil {
  1455. toAccumulate = NewSummaryAllocationSetRange()
  1456. as = as.Clone()
  1457. }
  1458. toAccumulate.Append(as)
  1459. sas, err := toAccumulate.accumulate()
  1460. if err != nil {
  1461. return nil, fmt.Errorf("error accumulating result: %s", err)
  1462. }
  1463. toAccumulate = NewSummaryAllocationSetRange(sas)
  1464. // current assumption is the week always ends on Saturday, or when there are no more summary allocation sets
  1465. if dayOfWeek == time.Saturday || i == len(sasr.SummaryAllocationSets)-1 {
  1466. if length := len(toAccumulate.SummaryAllocationSets); length != 1 {
  1467. return nil, fmt.Errorf("failed accumulation, detected %d sets instead of 1", length)
  1468. }
  1469. result.Append(toAccumulate.SummaryAllocationSets[0])
  1470. toAccumulate = nil
  1471. }
  1472. }
  1473. return result, nil
  1474. }
  1475. func (sasr *SummaryAllocationSetRange) Clone() *SummaryAllocationSetRange {
  1476. return sasr.clone()
  1477. }
  1478. // clone returns a new SummaryAllocationSetRange cloned from the existing SASR
  1479. func (sasr *SummaryAllocationSetRange) clone() *SummaryAllocationSetRange {
  1480. sasrSource := NewSummaryAllocationSetRange()
  1481. sasrSource.Window = sasr.Window.Clone()
  1482. sasrSource.Step = sasr.Step
  1483. sasrSource.Message = sasr.Message
  1484. for _, sas := range sasr.SummaryAllocationSets {
  1485. var sasClone *SummaryAllocationSet = nil
  1486. if sas != nil {
  1487. sasClone = sas.Clone()
  1488. }
  1489. sasrSource.Append(sasClone)
  1490. }
  1491. return sasrSource
  1492. }