2
0

aggregation.go 68 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826
  1. package costmodel
  2. import (
  3. "fmt"
  4. "math"
  5. "net/http"
  6. "regexp"
  7. "sort"
  8. "strconv"
  9. "strings"
  10. "time"
  11. "github.com/julienschmidt/httprouter"
  12. "github.com/kubecost/cost-model/pkg/cloud"
  13. "github.com/kubecost/cost-model/pkg/env"
  14. "github.com/kubecost/cost-model/pkg/kubecost"
  15. "github.com/kubecost/cost-model/pkg/log"
  16. "github.com/kubecost/cost-model/pkg/prom"
  17. "github.com/kubecost/cost-model/pkg/thanos"
  18. "github.com/kubecost/cost-model/pkg/util"
  19. "github.com/patrickmn/go-cache"
  20. prometheusClient "github.com/prometheus/client_golang/api"
  21. "k8s.io/klog"
  22. )
  23. const (
  24. // SplitTypeWeighted signals that shared costs should be shared
  25. // proportionally, rather than evenly
  26. SplitTypeWeighted = "weighted"
  27. // UnallocatedSubfield indicates an allocation datum that does not have the
  28. // chosen Aggregator; e.g. during aggregation by some label, there may be
  29. // cost data that do not have the given label.
  30. UnallocatedSubfield = "__unallocated__"
  31. )
  32. // Aggregation describes aggregated cost data, containing cumulative cost and
  33. // allocation data per resource, vectors of rate data per resource, efficiency
  34. // data, and metadata describing the type of aggregation operation.
  35. type Aggregation struct {
  36. Aggregator string `json:"aggregation"`
  37. Subfields []string `json:"subfields,omitempty"`
  38. Environment string `json:"environment"`
  39. Cluster string `json:"cluster,omitempty"`
  40. Properties *kubecost.Properties `json:"-"`
  41. CPUAllocationHourlyAverage float64 `json:"cpuAllocationAverage"`
  42. CPUAllocationVectors []*util.Vector `json:"-"`
  43. CPUAllocationTotal float64 `json:"-"`
  44. CPUCost float64 `json:"cpuCost"`
  45. CPUCostVector []*util.Vector `json:"cpuCostVector,omitempty"`
  46. CPUEfficiency float64 `json:"cpuEfficiency"`
  47. CPURequestedVectors []*util.Vector `json:"-"`
  48. CPUUsedVectors []*util.Vector `json:"-"`
  49. Efficiency float64 `json:"efficiency"`
  50. GPUAllocationHourlyAverage float64 `json:"gpuAllocationAverage"`
  51. GPUAllocationVectors []*util.Vector `json:"-"`
  52. GPUCost float64 `json:"gpuCost"`
  53. GPUCostVector []*util.Vector `json:"gpuCostVector,omitempty"`
  54. GPUAllocationTotal float64 `json:"-"`
  55. RAMAllocationHourlyAverage float64 `json:"ramAllocationAverage"`
  56. RAMAllocationVectors []*util.Vector `json:"-"`
  57. RAMAllocationTotal float64 `json:"-"`
  58. RAMCost float64 `json:"ramCost"`
  59. RAMCostVector []*util.Vector `json:"ramCostVector,omitempty"`
  60. RAMEfficiency float64 `json:"ramEfficiency"`
  61. RAMRequestedVectors []*util.Vector `json:"-"`
  62. RAMUsedVectors []*util.Vector `json:"-"`
  63. PVAllocationHourlyAverage float64 `json:"pvAllocationAverage"`
  64. PVAllocationVectors []*util.Vector `json:"-"`
  65. PVAllocationTotal float64 `json:"-"`
  66. PVCost float64 `json:"pvCost"`
  67. PVCostVector []*util.Vector `json:"pvCostVector,omitempty"`
  68. NetworkCost float64 `json:"networkCost"`
  69. NetworkCostVector []*util.Vector `json:"networkCostVector,omitempty"`
  70. SharedCost float64 `json:"sharedCost"`
  71. TotalCost float64 `json:"totalCost"`
  72. TotalCostVector []*util.Vector `json:"totalCostVector,omitempty"`
  73. }
  74. // TotalHours determines the amount of hours the Aggregation covers, as a
  75. // function of the cost vectors and the resolution of those vectors' data
  76. func (a *Aggregation) TotalHours(resolutionHours float64) float64 {
  77. length := 1
  78. if length < len(a.CPUCostVector) {
  79. length = len(a.CPUCostVector)
  80. }
  81. if length < len(a.RAMCostVector) {
  82. length = len(a.RAMCostVector)
  83. }
  84. if length < len(a.PVCostVector) {
  85. length = len(a.PVCostVector)
  86. }
  87. if length < len(a.GPUCostVector) {
  88. length = len(a.GPUCostVector)
  89. }
  90. if length < len(a.NetworkCostVector) {
  91. length = len(a.NetworkCostVector)
  92. }
  93. return float64(length) * resolutionHours
  94. }
  95. // RateCoefficient computes the coefficient by which the total cost needs to be
  96. // multiplied in order to convert totals costs into per-rate costs.
  97. func (a *Aggregation) RateCoefficient(rateStr string, resolutionHours float64) float64 {
  98. // monthly rate = (730.0)*(total cost)/(total hours)
  99. // daily rate = (24.0)*(total cost)/(total hours)
  100. // hourly rate = (1.0)*(total cost)/(total hours)
  101. // default to hourly rate
  102. coeff := 1.0
  103. switch rateStr {
  104. case "daily":
  105. coeff = util.HoursPerDay
  106. case "monthly":
  107. coeff = util.HoursPerMonth
  108. }
  109. return coeff / a.TotalHours(resolutionHours)
  110. }
  111. type SharedResourceInfo struct {
  112. ShareResources bool
  113. SharedNamespace map[string]bool
  114. LabelSelectors map[string]map[string]bool
  115. }
  116. type SharedCostInfo struct {
  117. Name string
  118. Cost float64
  119. ShareType string
  120. }
  121. func (s *SharedResourceInfo) IsSharedResource(costDatum *CostData) bool {
  122. // exists in a shared namespace
  123. if _, ok := s.SharedNamespace[costDatum.Namespace]; ok {
  124. return true
  125. }
  126. // has at least one shared label (OR, not AND in the case of multiple labels)
  127. for labelName, labelValues := range s.LabelSelectors {
  128. if val, ok := costDatum.Labels[labelName]; ok && labelValues[val] {
  129. return true
  130. }
  131. }
  132. return false
  133. }
  134. func NewSharedResourceInfo(shareResources bool, sharedNamespaces []string, labelNames []string, labelValues []string) *SharedResourceInfo {
  135. sr := &SharedResourceInfo{
  136. ShareResources: shareResources,
  137. SharedNamespace: make(map[string]bool),
  138. LabelSelectors: make(map[string]map[string]bool),
  139. }
  140. for _, ns := range sharedNamespaces {
  141. sr.SharedNamespace[strings.Trim(ns, " ")] = true
  142. }
  143. // Creating a map of label name to label value, but only if
  144. // the cardinality matches
  145. if len(labelNames) == len(labelValues) {
  146. for i := range labelNames {
  147. cleanedLname := prom.SanitizeLabelName(strings.Trim(labelNames[i], " "))
  148. if values, ok := sr.LabelSelectors[cleanedLname]; ok {
  149. values[strings.Trim(labelValues[i], " ")] = true
  150. } else {
  151. sr.LabelSelectors[cleanedLname] = map[string]bool{strings.Trim(labelValues[i], " "): true}
  152. }
  153. }
  154. }
  155. return sr
  156. }
  157. func GetTotalContainerCost(costData map[string]*CostData, rate string, cp cloud.Provider, discount float64, customDiscount float64, idleCoefficients map[string]float64) float64 {
  158. totalContainerCost := 0.0
  159. for _, costDatum := range costData {
  160. clusterID := costDatum.ClusterID
  161. cpuv, ramv, gpuv, pvvs, netv := getPriceVectors(cp, costDatum, rate, discount, customDiscount, idleCoefficients[clusterID])
  162. totalContainerCost += totalVectors(cpuv)
  163. totalContainerCost += totalVectors(ramv)
  164. totalContainerCost += totalVectors(gpuv)
  165. for _, pv := range pvvs {
  166. totalContainerCost += totalVectors(pv)
  167. }
  168. totalContainerCost += totalVectors(netv)
  169. }
  170. return totalContainerCost
  171. }
  172. func (a *Accesses) ComputeIdleCoefficient(costData map[string]*CostData, cli prometheusClient.Client, cp cloud.Provider, discount float64, customDiscount float64, windowString, offset string) (map[string]float64, error) {
  173. coefficients := make(map[string]float64)
  174. profileName := "ComputeIdleCoefficient: ComputeClusterCosts"
  175. profileStart := time.Now()
  176. var clusterCosts map[string]*ClusterCosts
  177. var err error
  178. key := fmt.Sprintf("%s:%s", windowString, offset)
  179. if data, valid := a.ClusterCostsCache.Get(key); valid {
  180. clusterCosts = data.(map[string]*ClusterCosts)
  181. } else {
  182. clusterCosts, err = a.ComputeClusterCosts(cli, cp, windowString, offset, false)
  183. if err != nil {
  184. return nil, err
  185. }
  186. }
  187. measureTime(profileStart, profileThreshold, profileName)
  188. for cid, costs := range clusterCosts {
  189. if costs.CPUCumulative == 0 && costs.RAMCumulative == 0 && costs.StorageCumulative == 0 {
  190. klog.V(1).Infof("[Warning] No ClusterCosts data for cluster '%s'. Is it emitting data?", cid)
  191. coefficients[cid] = 1.0
  192. continue
  193. }
  194. if costs.TotalCumulative == 0 {
  195. return nil, fmt.Errorf("TotalCumulative cluster cost for cluster '%s' returned 0 over window '%s' offset '%s'", cid, windowString, offset)
  196. }
  197. totalContainerCost := 0.0
  198. for _, costDatum := range costData {
  199. if costDatum.ClusterID == cid {
  200. cpuv, ramv, gpuv, pvvs, _ := getPriceVectors(cp, costDatum, "", discount, customDiscount, 1)
  201. totalContainerCost += totalVectors(cpuv)
  202. totalContainerCost += totalVectors(ramv)
  203. totalContainerCost += totalVectors(gpuv)
  204. for _, pv := range pvvs {
  205. totalContainerCost += totalVectors(pv)
  206. }
  207. }
  208. }
  209. coeff := totalContainerCost / costs.TotalCumulative
  210. coefficients[cid] = coeff
  211. }
  212. return coefficients, nil
  213. }
  214. // AggregationOptions provides optional parameters to AggregateCostData, allowing callers to perform more complex operations
  215. type AggregationOptions struct {
  216. Discount float64 // percent by which to discount CPU, RAM, and GPU cost
  217. CustomDiscount float64 // additional custom discount applied to all prices
  218. IdleCoefficients map[string]float64 // scales costs by amount of idle resources on a per-cluster basis
  219. IncludeEfficiency bool // set to true to receive efficiency/usage data
  220. IncludeTimeSeries bool // set to true to receive time series data
  221. Rate string // set to "hourly", "daily", or "monthly" to receive cost rate, rather than cumulative cost
  222. ResolutionHours float64
  223. SharedResourceInfo *SharedResourceInfo
  224. SharedCosts map[string]*SharedCostInfo
  225. FilteredContainerCount int
  226. FilteredEnvironments map[string]int
  227. SharedSplit string
  228. TotalContainerCost float64
  229. }
  230. // Helper method to test request/usgae values against allocation averages for efficiency scores. Generate a warning log if
  231. // clamp is required
  232. func clampAverage(requestsAvg float64, usedAverage float64, allocationAvg float64, resource string) (float64, float64) {
  233. rAvg := requestsAvg
  234. if rAvg > allocationAvg {
  235. klog.V(4).Infof("[Warning] Average %s Requested (%f) > Average %s Allocated (%f). Clamping.", resource, rAvg, resource, allocationAvg)
  236. rAvg = allocationAvg
  237. }
  238. uAvg := usedAverage
  239. if uAvg > allocationAvg {
  240. klog.V(4).Infof("[Warning]: Average %s Used (%f) > Average %s Allocated (%f). Clamping.", resource, uAvg, resource, allocationAvg)
  241. uAvg = allocationAvg
  242. }
  243. return rAvg, uAvg
  244. }
  245. // AggregateCostData aggregates raw cost data by field; e.g. namespace, cluster, service, or label. In the case of label, callers
  246. // must pass a slice of subfields indicating the labels by which to group. Provider is used to define custom resource pricing.
  247. // See AggregationOptions for optional parameters.
  248. func AggregateCostData(costData map[string]*CostData, field string, subfields []string, cp cloud.Provider, opts *AggregationOptions) map[string]*Aggregation {
  249. discount := opts.Discount
  250. customDiscount := opts.CustomDiscount
  251. idleCoefficients := opts.IdleCoefficients
  252. includeTimeSeries := opts.IncludeTimeSeries
  253. includeEfficiency := opts.IncludeEfficiency
  254. rate := opts.Rate
  255. sr := opts.SharedResourceInfo
  256. resolutionHours := 1.0
  257. if opts.ResolutionHours > 0.0 {
  258. resolutionHours = opts.ResolutionHours
  259. }
  260. if idleCoefficients == nil {
  261. idleCoefficients = make(map[string]float64)
  262. }
  263. // aggregations collects key-value pairs of resource group-to-aggregated data
  264. // e.g. namespace-to-data or label-value-to-data
  265. aggregations := make(map[string]*Aggregation)
  266. // sharedResourceCost is the running total cost of resources that should be reported
  267. // as shared across all other resources, rather than reported as a stand-alone category
  268. sharedResourceCost := 0.0
  269. for _, costDatum := range costData {
  270. idleCoefficient, ok := idleCoefficients[costDatum.ClusterID]
  271. if !ok {
  272. idleCoefficient = 1.0
  273. }
  274. if sr != nil && sr.ShareResources && sr.IsSharedResource(costDatum) {
  275. cpuv, ramv, gpuv, pvvs, netv := getPriceVectors(cp, costDatum, rate, discount, customDiscount, idleCoefficient)
  276. sharedResourceCost += totalVectors(cpuv)
  277. sharedResourceCost += totalVectors(ramv)
  278. sharedResourceCost += totalVectors(gpuv)
  279. sharedResourceCost += totalVectors(netv)
  280. for _, pv := range pvvs {
  281. sharedResourceCost += totalVectors(pv)
  282. }
  283. } else {
  284. if field == "cluster" {
  285. aggregateDatum(cp, aggregations, costDatum, field, subfields, rate, costDatum.ClusterID, discount, customDiscount, idleCoefficient, false)
  286. } else if field == "node" {
  287. aggregateDatum(cp, aggregations, costDatum, field, subfields, rate, costDatum.NodeName, discount, customDiscount, idleCoefficient, false)
  288. } else if field == "namespace" {
  289. aggregateDatum(cp, aggregations, costDatum, field, subfields, rate, costDatum.Namespace, discount, customDiscount, idleCoefficient, false)
  290. } else if field == "service" {
  291. if len(costDatum.Services) > 0 {
  292. aggregateDatum(cp, aggregations, costDatum, field, subfields, rate, costDatum.Namespace+"/"+costDatum.Services[0], discount, customDiscount, idleCoefficient, false)
  293. } else {
  294. aggregateDatum(cp, aggregations, costDatum, field, subfields, rate, UnallocatedSubfield, discount, customDiscount, idleCoefficient, false)
  295. }
  296. } else if field == "deployment" {
  297. if len(costDatum.Deployments) > 0 {
  298. aggregateDatum(cp, aggregations, costDatum, field, subfields, rate, costDatum.Namespace+"/"+costDatum.Deployments[0], discount, customDiscount, idleCoefficient, false)
  299. } else {
  300. aggregateDatum(cp, aggregations, costDatum, field, subfields, rate, UnallocatedSubfield, discount, customDiscount, idleCoefficient, false)
  301. }
  302. } else if field == "statefulset" {
  303. if len(costDatum.Statefulsets) > 0 {
  304. aggregateDatum(cp, aggregations, costDatum, field, subfields, rate, costDatum.Namespace+"/"+costDatum.Statefulsets[0], discount, customDiscount, idleCoefficient, false)
  305. } else {
  306. aggregateDatum(cp, aggregations, costDatum, field, subfields, rate, UnallocatedSubfield, discount, customDiscount, idleCoefficient, false)
  307. }
  308. } else if field == "daemonset" {
  309. if len(costDatum.Daemonsets) > 0 {
  310. aggregateDatum(cp, aggregations, costDatum, field, subfields, rate, costDatum.Namespace+"/"+costDatum.Daemonsets[0], discount, customDiscount, idleCoefficient, false)
  311. } else {
  312. aggregateDatum(cp, aggregations, costDatum, field, subfields, rate, UnallocatedSubfield, discount, customDiscount, idleCoefficient, false)
  313. }
  314. } else if field == "controller" {
  315. if controller, kind, hasController := costDatum.GetController(); hasController {
  316. key := fmt.Sprintf("%s/%s:%s", costDatum.Namespace, kind, controller)
  317. aggregateDatum(cp, aggregations, costDatum, field, subfields, rate, key, discount, customDiscount, idleCoefficient, false)
  318. } else {
  319. aggregateDatum(cp, aggregations, costDatum, field, subfields, rate, UnallocatedSubfield, discount, customDiscount, idleCoefficient, false)
  320. }
  321. } else if field == "label" {
  322. found := false
  323. if costDatum.Labels != nil {
  324. for _, sf := range subfields {
  325. if subfieldName, ok := costDatum.Labels[sf]; ok {
  326. aggregateDatum(cp, aggregations, costDatum, field, subfields, rate, subfieldName, discount, customDiscount, idleCoefficient, false)
  327. found = true
  328. break
  329. }
  330. }
  331. }
  332. if !found {
  333. aggregateDatum(cp, aggregations, costDatum, field, subfields, rate, UnallocatedSubfield, discount, customDiscount, idleCoefficient, false)
  334. }
  335. } else if field == "pod" {
  336. aggregateDatum(cp, aggregations, costDatum, field, subfields, rate, costDatum.Namespace+"/"+costDatum.PodName, discount, customDiscount, idleCoefficient, false)
  337. } else if field == "container" {
  338. key := fmt.Sprintf("%s/%s/%s/%s", costDatum.ClusterID, costDatum.Namespace, costDatum.PodName, costDatum.Name)
  339. aggregateDatum(cp, aggregations, costDatum, field, subfields, rate, key, discount, customDiscount, idleCoefficient, true)
  340. }
  341. }
  342. }
  343. for key, agg := range aggregations {
  344. sharedCoefficient := 1 / float64(len(opts.FilteredEnvironments)+len(aggregations))
  345. agg.CPUCost = totalVectors(agg.CPUCostVector)
  346. agg.RAMCost = totalVectors(agg.RAMCostVector)
  347. agg.GPUCost = totalVectors(agg.GPUCostVector)
  348. agg.PVCost = totalVectors(agg.PVCostVector)
  349. agg.NetworkCost = totalVectors(agg.NetworkCostVector)
  350. if opts.SharedSplit == SplitTypeWeighted {
  351. d := opts.TotalContainerCost - sharedResourceCost
  352. if d == 0 {
  353. klog.V(1).Infof("[Warning] Total container cost '%f' and shared resource cost '%f are the same'. Setting sharedCoefficient to 1", opts.TotalContainerCost, sharedResourceCost)
  354. sharedCoefficient = 1.0
  355. } else {
  356. sharedCoefficient = (agg.CPUCost + agg.RAMCost + agg.GPUCost + agg.PVCost + agg.NetworkCost) / d
  357. }
  358. }
  359. agg.SharedCost = sharedResourceCost * sharedCoefficient
  360. for _, v := range opts.SharedCosts {
  361. agg.SharedCost += v.Cost * sharedCoefficient
  362. }
  363. if rate != "" {
  364. rateCoeff := agg.RateCoefficient(rate, resolutionHours)
  365. agg.CPUCost *= rateCoeff
  366. agg.RAMCost *= rateCoeff
  367. agg.GPUCost *= rateCoeff
  368. agg.PVCost *= rateCoeff
  369. agg.NetworkCost *= rateCoeff
  370. agg.SharedCost *= rateCoeff
  371. }
  372. agg.TotalCost = agg.CPUCost + agg.RAMCost + agg.GPUCost + agg.PVCost + agg.NetworkCost + agg.SharedCost
  373. // Evicted and Completed Pods can still show up here, but have 0 cost.
  374. // Filter these by default. Any reason to keep them?
  375. if agg.TotalCost == 0 {
  376. delete(aggregations, key)
  377. continue
  378. }
  379. // CPU, RAM, and PV allocation are cumulative per-datum, whereas GPU is rate per-datum
  380. agg.CPUAllocationHourlyAverage = totalVectors(agg.CPUAllocationVectors) / agg.TotalHours(resolutionHours)
  381. agg.RAMAllocationHourlyAverage = totalVectors(agg.RAMAllocationVectors) / agg.TotalHours(resolutionHours)
  382. agg.GPUAllocationHourlyAverage = averageVectors(agg.GPUAllocationVectors)
  383. agg.PVAllocationHourlyAverage = totalVectors(agg.PVAllocationVectors) / agg.TotalHours(resolutionHours)
  384. // TODO niko/etl does this check out for GPU data? Do we need to rewrite GPU queries to be
  385. // culumative?
  386. agg.CPUAllocationTotal = totalVectors(agg.CPUAllocationVectors)
  387. agg.GPUAllocationTotal = totalVectors(agg.GPUAllocationVectors)
  388. agg.PVAllocationTotal = totalVectors(agg.PVAllocationVectors)
  389. agg.RAMAllocationTotal = totalVectors(agg.RAMAllocationVectors)
  390. if includeEfficiency {
  391. // Default both RAM and CPU to 0% efficiency so that a 0-requested, 0-allocated, 0-used situation
  392. // returns 0% efficiency, which should be a red-flag.
  393. //
  394. // If non-zero numbers are available, then efficiency is defined as:
  395. // idlePercentage = (requested - used) / allocated
  396. // efficiency = (1.0 - idlePercentage)
  397. //
  398. // It is possible to score > 100% efficiency, which is meant to be interpreted as a red flag.
  399. // It is not possible to score < 0% efficiency.
  400. agg.CPUEfficiency = 0.0
  401. CPUIdle := 0.0
  402. if agg.CPUAllocationHourlyAverage > 0.0 {
  403. avgCPURequested := averageVectors(agg.CPURequestedVectors)
  404. avgCPUUsed := averageVectors(agg.CPUUsedVectors)
  405. // Clamp averages, log range violations
  406. avgCPURequested, avgCPUUsed = clampAverage(avgCPURequested, avgCPUUsed, agg.CPUAllocationHourlyAverage, "CPU")
  407. CPUIdle = ((avgCPURequested - avgCPUUsed) / agg.CPUAllocationHourlyAverage)
  408. agg.CPUEfficiency = 1.0 - CPUIdle
  409. }
  410. agg.RAMEfficiency = 0.0
  411. RAMIdle := 0.0
  412. if agg.RAMAllocationHourlyAverage > 0.0 {
  413. avgRAMRequested := averageVectors(agg.RAMRequestedVectors)
  414. avgRAMUsed := averageVectors(agg.RAMUsedVectors)
  415. // Clamp averages, log range violations
  416. avgRAMRequested, avgRAMUsed = clampAverage(avgRAMRequested, avgRAMUsed, agg.RAMAllocationHourlyAverage, "RAM")
  417. RAMIdle = ((avgRAMRequested - avgRAMUsed) / agg.RAMAllocationHourlyAverage)
  418. agg.RAMEfficiency = 1.0 - RAMIdle
  419. }
  420. // Score total efficiency by the sum of CPU and RAM efficiency, weighted by their
  421. // respective total costs.
  422. agg.Efficiency = 0.0
  423. if (agg.CPUCost + agg.RAMCost) > 0 {
  424. agg.Efficiency = ((agg.CPUCost * agg.CPUEfficiency) + (agg.RAMCost * agg.RAMEfficiency)) / (agg.CPUCost + agg.RAMCost)
  425. }
  426. }
  427. // convert RAM from bytes to GiB
  428. agg.RAMAllocationHourlyAverage = agg.RAMAllocationHourlyAverage / 1024 / 1024 / 1024
  429. // convert storage from bytes to GiB
  430. agg.PVAllocationHourlyAverage = agg.PVAllocationHourlyAverage / 1024 / 1024 / 1024
  431. // remove time series data if it is not explicitly requested
  432. if !includeTimeSeries {
  433. agg.CPUCostVector = nil
  434. agg.RAMCostVector = nil
  435. agg.GPUCostVector = nil
  436. agg.PVCostVector = nil
  437. agg.NetworkCostVector = nil
  438. agg.TotalCostVector = nil
  439. } else { // otherwise compute a totalcostvector
  440. v1 := addVectors(agg.CPUCostVector, agg.RAMCostVector)
  441. v2 := addVectors(v1, agg.GPUCostVector)
  442. v3 := addVectors(v2, agg.PVCostVector)
  443. v4 := addVectors(v3, agg.NetworkCostVector)
  444. agg.TotalCostVector = v4
  445. }
  446. // Typesafety checks
  447. if math.IsNaN(agg.CPUAllocationHourlyAverage) || math.IsInf(agg.CPUAllocationHourlyAverage, 0) {
  448. klog.V(1).Infof("[Warning] CPUAllocationHourlyAverage is %f for '%s: %s/%s'", agg.CPUAllocationHourlyAverage, agg.Cluster, agg.Aggregator, agg.Environment)
  449. agg.CPUAllocationHourlyAverage = 0
  450. }
  451. if math.IsNaN(agg.CPUCost) || math.IsInf(agg.CPUCost, 0) {
  452. klog.V(1).Infof("[Warning] CPUCost is %f for '%s: %s/%s'", agg.CPUCost, agg.Cluster, agg.Aggregator, agg.Environment)
  453. agg.CPUCost = 0
  454. }
  455. if math.IsNaN(agg.CPUEfficiency) || math.IsInf(agg.CPUEfficiency, 0) {
  456. klog.V(1).Infof("[Warning] CPUEfficiency is %f for '%s: %s/%s'", agg.CPUEfficiency, agg.Cluster, agg.Aggregator, agg.Environment)
  457. agg.CPUEfficiency = 0
  458. }
  459. if math.IsNaN(agg.Efficiency) || math.IsInf(agg.Efficiency, 0) {
  460. klog.V(1).Infof("[Warning] Efficiency is %f for '%s: %s/%s'", agg.Efficiency, agg.Cluster, agg.Aggregator, agg.Environment)
  461. agg.Efficiency = 0
  462. }
  463. if math.IsNaN(agg.GPUAllocationHourlyAverage) || math.IsInf(agg.GPUAllocationHourlyAverage, 0) {
  464. klog.V(1).Infof("[Warning] GPUAllocationHourlyAverage is %f for '%s: %s/%s'", agg.GPUAllocationHourlyAverage, agg.Cluster, agg.Aggregator, agg.Environment)
  465. agg.GPUAllocationHourlyAverage = 0
  466. }
  467. if math.IsNaN(agg.GPUCost) || math.IsInf(agg.GPUCost, 0) {
  468. klog.V(1).Infof("[Warning] GPUCost is %f for '%s: %s/%s'", agg.GPUCost, agg.Cluster, agg.Aggregator, agg.Environment)
  469. agg.GPUCost = 0
  470. }
  471. if math.IsNaN(agg.RAMAllocationHourlyAverage) || math.IsInf(agg.RAMAllocationHourlyAverage, 0) {
  472. klog.V(1).Infof("[Warning] RAMAllocationHourlyAverage is %f for '%s: %s/%s'", agg.RAMAllocationHourlyAverage, agg.Cluster, agg.Aggregator, agg.Environment)
  473. agg.RAMAllocationHourlyAverage = 0
  474. }
  475. if math.IsNaN(agg.RAMCost) || math.IsInf(agg.RAMCost, 0) {
  476. klog.V(1).Infof("[Warning] RAMCost is %f for '%s: %s/%s'", agg.RAMCost, agg.Cluster, agg.Aggregator, agg.Environment)
  477. agg.RAMCost = 0
  478. }
  479. if math.IsNaN(agg.RAMEfficiency) || math.IsInf(agg.RAMEfficiency, 0) {
  480. klog.V(1).Infof("[Warning] RAMEfficiency is %f for '%s: %s/%s'", agg.RAMEfficiency, agg.Cluster, agg.Aggregator, agg.Environment)
  481. agg.RAMEfficiency = 0
  482. }
  483. if math.IsNaN(agg.PVAllocationHourlyAverage) || math.IsInf(agg.PVAllocationHourlyAverage, 0) {
  484. klog.V(1).Infof("[Warning] PVAllocationHourlyAverage is %f for '%s: %s/%s'", agg.PVAllocationHourlyAverage, agg.Cluster, agg.Aggregator, agg.Environment)
  485. agg.PVAllocationHourlyAverage = 0
  486. }
  487. if math.IsNaN(agg.PVCost) || math.IsInf(agg.PVCost, 0) {
  488. klog.V(1).Infof("[Warning] PVCost is %f for '%s: %s/%s'", agg.PVCost, agg.Cluster, agg.Aggregator, agg.Environment)
  489. agg.PVCost = 0
  490. }
  491. if math.IsNaN(agg.NetworkCost) || math.IsInf(agg.NetworkCost, 0) {
  492. klog.V(1).Infof("[Warning] NetworkCost is %f for '%s: %s/%s'", agg.NetworkCost, agg.Cluster, agg.Aggregator, agg.Environment)
  493. agg.NetworkCost = 0
  494. }
  495. if math.IsNaN(agg.SharedCost) || math.IsInf(agg.SharedCost, 0) {
  496. klog.V(1).Infof("[Warning] SharedCost is %f for '%s: %s/%s'", agg.SharedCost, agg.Cluster, agg.Aggregator, agg.Environment)
  497. agg.SharedCost = 0
  498. }
  499. if math.IsNaN(agg.TotalCost) || math.IsInf(agg.TotalCost, 0) {
  500. klog.V(1).Infof("[Warning] TotalCost is %f for '%s: %s/%s'", agg.TotalCost, agg.Cluster, agg.Aggregator, agg.Environment)
  501. agg.TotalCost = 0
  502. }
  503. }
  504. return aggregations
  505. }
  506. func aggregateDatum(cp cloud.Provider, aggregations map[string]*Aggregation, costDatum *CostData, field string, subfields []string, rate string, key string, discount float64, customDiscount float64, idleCoefficient float64, includeProperties bool) {
  507. // add new entry to aggregation results if a new key is encountered
  508. if _, ok := aggregations[key]; !ok {
  509. agg := &Aggregation{
  510. Aggregator: field,
  511. Environment: key,
  512. }
  513. if len(subfields) > 0 {
  514. agg.Subfields = subfields
  515. }
  516. if includeProperties {
  517. props := &kubecost.Properties{}
  518. props.SetCluster(costDatum.ClusterID)
  519. props.SetNode(costDatum.NodeName)
  520. if controller, kind, hasController := costDatum.GetController(); hasController {
  521. props.SetController(controller)
  522. props.SetControllerKind(kind)
  523. }
  524. props.SetLabels(costDatum.Labels)
  525. props.SetNamespace(costDatum.Namespace)
  526. props.SetPod(costDatum.PodName)
  527. props.SetServices(costDatum.Services)
  528. props.SetContainer(costDatum.Name)
  529. agg.Properties = props
  530. }
  531. aggregations[key] = agg
  532. }
  533. mergeVectors(cp, costDatum, aggregations[key], rate, discount, customDiscount, idleCoefficient)
  534. }
  535. func mergeVectors(cp cloud.Provider, costDatum *CostData, aggregation *Aggregation, rate string, discount float64, customDiscount float64, idleCoefficient float64) {
  536. aggregation.CPUAllocationVectors = addVectors(costDatum.CPUAllocation, aggregation.CPUAllocationVectors)
  537. aggregation.CPURequestedVectors = addVectors(costDatum.CPUReq, aggregation.CPURequestedVectors)
  538. aggregation.CPUUsedVectors = addVectors(costDatum.CPUUsed, aggregation.CPUUsedVectors)
  539. aggregation.RAMAllocationVectors = addVectors(costDatum.RAMAllocation, aggregation.RAMAllocationVectors)
  540. aggregation.RAMRequestedVectors = addVectors(costDatum.RAMReq, aggregation.RAMRequestedVectors)
  541. aggregation.RAMUsedVectors = addVectors(costDatum.RAMUsed, aggregation.RAMUsedVectors)
  542. aggregation.GPUAllocationVectors = addVectors(costDatum.GPUReq, aggregation.GPUAllocationVectors)
  543. for _, pvcd := range costDatum.PVCData {
  544. aggregation.PVAllocationVectors = addVectors(pvcd.Values, aggregation.PVAllocationVectors)
  545. }
  546. cpuv, ramv, gpuv, pvvs, netv := getPriceVectors(cp, costDatum, rate, discount, customDiscount, idleCoefficient)
  547. aggregation.CPUCostVector = addVectors(cpuv, aggregation.CPUCostVector)
  548. aggregation.RAMCostVector = addVectors(ramv, aggregation.RAMCostVector)
  549. aggregation.GPUCostVector = addVectors(gpuv, aggregation.GPUCostVector)
  550. aggregation.NetworkCostVector = addVectors(netv, aggregation.NetworkCostVector)
  551. for _, vectorList := range pvvs {
  552. aggregation.PVCostVector = addVectors(aggregation.PVCostVector, vectorList)
  553. }
  554. }
  555. // Returns the blended discounts applied to the node as a result of global discounts and reserved instance
  556. // discounts
  557. func getDiscounts(costDatum *CostData, cpuCost float64, ramCost float64, discount float64) (float64, float64) {
  558. if costDatum.NodeData == nil {
  559. return discount, discount
  560. }
  561. if costDatum.NodeData.IsSpot() {
  562. return 0, 0
  563. }
  564. reserved := costDatum.NodeData.Reserved
  565. // blended discounts
  566. blendedCPUDiscount := discount
  567. blendedRAMDiscount := discount
  568. if reserved != nil && reserved.CPUCost > 0 && reserved.RAMCost > 0 {
  569. reservedCPUDiscount := 0.0
  570. if cpuCost == 0 {
  571. klog.V(1).Infof("[Warning] No cpu cost found for cluster '%s' node '%s'", costDatum.ClusterID, costDatum.NodeName)
  572. } else {
  573. reservedCPUDiscount = 1.0 - (reserved.CPUCost / cpuCost)
  574. }
  575. reservedRAMDiscount := 0.0
  576. if ramCost == 0 {
  577. klog.V(1).Infof("[Warning] No ram cost found for cluster '%s' node '%s'", costDatum.ClusterID, costDatum.NodeName)
  578. } else {
  579. reservedRAMDiscount = 1.0 - (reserved.RAMCost / ramCost)
  580. }
  581. // AWS passes the # of reserved CPU and RAM as -1 to represent "All"
  582. if reserved.ReservedCPU < 0 && reserved.ReservedRAM < 0 {
  583. blendedCPUDiscount = reservedCPUDiscount
  584. blendedRAMDiscount = reservedRAMDiscount
  585. } else {
  586. nodeCPU, ierr := strconv.ParseInt(costDatum.NodeData.VCPU, 10, 64)
  587. nodeRAM, ferr := strconv.ParseFloat(costDatum.NodeData.RAMBytes, 64)
  588. if ierr == nil && ferr == nil {
  589. nodeRAMGB := nodeRAM / 1024 / 1024 / 1024
  590. reservedRAMGB := float64(reserved.ReservedRAM) / 1024 / 1024 / 1024
  591. nonReservedCPU := nodeCPU - reserved.ReservedCPU
  592. nonReservedRAM := nodeRAMGB - reservedRAMGB
  593. if nonReservedCPU == 0 {
  594. blendedCPUDiscount = reservedCPUDiscount
  595. } else {
  596. if nodeCPU == 0 {
  597. klog.V(1).Infof("[Warning] No ram found for cluster '%s' node '%s'", costDatum.ClusterID, costDatum.NodeName)
  598. } else {
  599. blendedCPUDiscount = (float64(reserved.ReservedCPU) * reservedCPUDiscount) + (float64(nonReservedCPU)*discount)/float64(nodeCPU)
  600. }
  601. }
  602. if nonReservedRAM == 0 {
  603. blendedRAMDiscount = reservedRAMDiscount
  604. } else {
  605. if nodeRAMGB == 0 {
  606. klog.V(1).Infof("[Warning] No ram found for cluster '%s' node '%s'", costDatum.ClusterID, costDatum.NodeName)
  607. } else {
  608. blendedRAMDiscount = (reservedRAMGB * reservedRAMDiscount) + (nonReservedRAM*discount)/nodeRAMGB
  609. }
  610. }
  611. }
  612. }
  613. }
  614. return blendedCPUDiscount, blendedRAMDiscount
  615. }
  616. func parseVectorPricing(cfg *cloud.CustomPricing, costDatum *CostData, cpuCostStr, ramCostStr, gpuCostStr, pvCostStr string) (float64, float64, float64, float64, bool) {
  617. usesCustom := false
  618. cpuCost, err := strconv.ParseFloat(cpuCostStr, 64)
  619. if err != nil || math.IsNaN(cpuCost) || math.IsInf(cpuCost, 0) || cpuCost == 0 {
  620. cpuCost, err = strconv.ParseFloat(cfg.CPU, 64)
  621. usesCustom = true
  622. if err != nil || math.IsNaN(cpuCost) || math.IsInf(cpuCost, 0) {
  623. cpuCost = 0
  624. }
  625. }
  626. ramCost, err := strconv.ParseFloat(ramCostStr, 64)
  627. if err != nil || math.IsNaN(ramCost) || math.IsInf(ramCost, 0) || ramCost == 0 {
  628. ramCost, err = strconv.ParseFloat(cfg.RAM, 64)
  629. usesCustom = true
  630. if err != nil || math.IsNaN(ramCost) || math.IsInf(ramCost, 0) {
  631. ramCost = 0
  632. }
  633. }
  634. gpuCost, err := strconv.ParseFloat(gpuCostStr, 64)
  635. if err != nil || math.IsNaN(gpuCost) || math.IsInf(gpuCost, 0) {
  636. gpuCost, err = strconv.ParseFloat(cfg.GPU, 64)
  637. if err != nil || math.IsNaN(gpuCost) || math.IsInf(gpuCost, 0) {
  638. gpuCost = 0
  639. }
  640. }
  641. pvCost, err := strconv.ParseFloat(pvCostStr, 64)
  642. if err != nil || math.IsNaN(cpuCost) || math.IsInf(cpuCost, 0) {
  643. pvCost, err = strconv.ParseFloat(cfg.Storage, 64)
  644. if err != nil || math.IsNaN(pvCost) || math.IsInf(pvCost, 0) {
  645. pvCost = 0
  646. }
  647. }
  648. return cpuCost, ramCost, gpuCost, pvCost, usesCustom
  649. }
  650. func getPriceVectors(cp cloud.Provider, costDatum *CostData, rate string, discount float64, customDiscount float64, idleCoefficient float64) ([]*util.Vector, []*util.Vector, []*util.Vector, [][]*util.Vector, []*util.Vector) {
  651. var cpuCost float64
  652. var ramCost float64
  653. var gpuCost float64
  654. var pvCost float64
  655. var usesCustom bool
  656. // If custom pricing is enabled and can be retrieved, replace
  657. // default cost values with custom values
  658. customPricing, err := cp.GetConfig()
  659. if err != nil {
  660. klog.Errorf("failed to load custom pricing: %s", err)
  661. }
  662. if cloud.CustomPricesEnabled(cp) && err == nil {
  663. var cpuCostStr string
  664. var ramCostStr string
  665. var gpuCostStr string
  666. var pvCostStr string
  667. if costDatum.NodeData.IsSpot() {
  668. cpuCostStr = customPricing.SpotCPU
  669. ramCostStr = customPricing.SpotRAM
  670. gpuCostStr = customPricing.SpotGPU
  671. } else {
  672. cpuCostStr = customPricing.CPU
  673. ramCostStr = customPricing.RAM
  674. gpuCostStr = customPricing.GPU
  675. }
  676. pvCostStr = customPricing.Storage
  677. cpuCost, ramCost, gpuCost, pvCost, usesCustom = parseVectorPricing(customPricing, costDatum, cpuCostStr, ramCostStr, gpuCostStr, pvCostStr)
  678. } else if costDatum.NodeData == nil && err == nil {
  679. cpuCostStr := customPricing.CPU
  680. ramCostStr := customPricing.RAM
  681. gpuCostStr := customPricing.GPU
  682. pvCostStr := customPricing.Storage
  683. cpuCost, ramCost, gpuCost, pvCost, usesCustom = parseVectorPricing(customPricing, costDatum, cpuCostStr, ramCostStr, gpuCostStr, pvCostStr)
  684. } else {
  685. cpuCostStr := costDatum.NodeData.VCPUCost
  686. ramCostStr := costDatum.NodeData.RAMCost
  687. gpuCostStr := costDatum.NodeData.GPUCost
  688. pvCostStr := costDatum.NodeData.StorageCost
  689. cpuCost, ramCost, gpuCost, pvCost, usesCustom = parseVectorPricing(customPricing, costDatum, cpuCostStr, ramCostStr, gpuCostStr, pvCostStr)
  690. }
  691. if usesCustom {
  692. log.DedupedWarningf(5, "No pricing data found for node `%s` , using custom pricing", costDatum.NodeName)
  693. }
  694. cpuDiscount, ramDiscount := getDiscounts(costDatum, cpuCost, ramCost, discount)
  695. klog.V(4).Infof("Node Name: %s", costDatum.NodeName)
  696. klog.V(4).Infof("Blended CPU Discount: %f", cpuDiscount)
  697. klog.V(4).Infof("Blended RAM Discount: %f", ramDiscount)
  698. // TODO should we try to apply the rate coefficient here or leave it as a totals-only metric?
  699. rateCoeff := 1.0
  700. if idleCoefficient == 0 {
  701. idleCoefficient = 1.0
  702. }
  703. cpuv := make([]*util.Vector, 0, len(costDatum.CPUAllocation))
  704. for _, val := range costDatum.CPUAllocation {
  705. cpuv = append(cpuv, &util.Vector{
  706. Timestamp: math.Round(val.Timestamp/10) * 10,
  707. Value: (val.Value * cpuCost * (1 - cpuDiscount) * (1 - customDiscount) / idleCoefficient) * rateCoeff,
  708. })
  709. }
  710. ramv := make([]*util.Vector, 0, len(costDatum.RAMAllocation))
  711. for _, val := range costDatum.RAMAllocation {
  712. ramv = append(ramv, &util.Vector{
  713. Timestamp: math.Round(val.Timestamp/10) * 10,
  714. Value: ((val.Value / 1024 / 1024 / 1024) * ramCost * (1 - ramDiscount) * (1 - customDiscount) / idleCoefficient) * rateCoeff,
  715. })
  716. }
  717. gpuv := make([]*util.Vector, 0, len(costDatum.GPUReq))
  718. for _, val := range costDatum.GPUReq {
  719. gpuv = append(gpuv, &util.Vector{
  720. Timestamp: math.Round(val.Timestamp/10) * 10,
  721. Value: (val.Value * gpuCost * (1 - discount) * (1 - customDiscount) / idleCoefficient) * rateCoeff,
  722. })
  723. }
  724. pvvs := make([][]*util.Vector, 0, len(costDatum.PVCData))
  725. for _, pvcData := range costDatum.PVCData {
  726. pvv := make([]*util.Vector, 0, len(pvcData.Values))
  727. if pvcData.Volume != nil {
  728. cost, _ := strconv.ParseFloat(pvcData.Volume.Cost, 64)
  729. // override with custom pricing if enabled
  730. if cloud.CustomPricesEnabled(cp) {
  731. cost = pvCost
  732. }
  733. for _, val := range pvcData.Values {
  734. pvv = append(pvv, &util.Vector{
  735. Timestamp: math.Round(val.Timestamp/10) * 10,
  736. Value: ((val.Value / 1024 / 1024 / 1024) * cost * (1 - customDiscount) / idleCoefficient) * rateCoeff,
  737. })
  738. }
  739. pvvs = append(pvvs, pvv)
  740. }
  741. }
  742. netv := make([]*util.Vector, 0, len(costDatum.NetworkData))
  743. for _, val := range costDatum.NetworkData {
  744. netv = append(netv, &util.Vector{
  745. Timestamp: math.Round(val.Timestamp/10) * 10,
  746. Value: val.Value,
  747. })
  748. }
  749. return cpuv, ramv, gpuv, pvvs, netv
  750. }
  751. func averageVectors(vectors []*util.Vector) float64 {
  752. if len(vectors) == 0 {
  753. return 0.0
  754. }
  755. return totalVectors(vectors) / float64(len(vectors))
  756. }
  757. func totalVectors(vectors []*util.Vector) float64 {
  758. total := 0.0
  759. for _, vector := range vectors {
  760. total += vector.Value
  761. }
  762. return total
  763. }
  764. // addVectors adds two slices of Vectors. Vector timestamps are rounded to the
  765. // nearest ten seconds to allow matching of Vectors within a delta allowance.
  766. // Matching Vectors are summed, while unmatched Vectors are passed through.
  767. // e.g. [(t=1, 1), (t=2, 2)] + [(t=2, 2), (t=3, 3)] = [(t=1, 1), (t=2, 4), (t=3, 3)]
  768. func addVectors(xvs []*util.Vector, yvs []*util.Vector) []*util.Vector {
  769. sumOp := func(result *util.Vector, x *float64, y *float64) bool {
  770. if x != nil && y != nil {
  771. result.Value = *x + *y
  772. } else if y != nil {
  773. result.Value = *y
  774. } else if x != nil {
  775. result.Value = *x
  776. }
  777. return true
  778. }
  779. return util.ApplyVectorOp(xvs, yvs, sumOp)
  780. }
  781. // minCostDataLength sets the minimum number of time series data required to
  782. // cache both raw and aggregated cost data
  783. const minCostDataLength = 2
  784. // EmptyDataError describes an error caused by empty cost data for some
  785. // defined interval
  786. type EmptyDataError struct {
  787. err error
  788. duration string
  789. offset string
  790. }
  791. // Error implements the error interface
  792. func (ede *EmptyDataError) Error() string {
  793. err := fmt.Sprintf("empty data for range: %s", ede.duration)
  794. if ede.offset != "" {
  795. err += fmt.Sprintf(" offset %s", ede.offset)
  796. }
  797. if ede.err != nil {
  798. err += fmt.Sprintf(": %s", ede.err)
  799. }
  800. return err
  801. }
  802. func costDataTimeSeriesLength(costData map[string]*CostData) int {
  803. l := 0
  804. for _, cd := range costData {
  805. if l < len(cd.RAMAllocation) {
  806. l = len(cd.RAMAllocation)
  807. }
  808. if l < len(cd.CPUAllocation) {
  809. l = len(cd.CPUAllocation)
  810. }
  811. }
  812. return l
  813. }
  814. // ScaleHourlyCostData converts per-hour cost data to per-resolution data. If the target resolution is higher (i.e. < 1.0h)
  815. // then we can do simple multiplication by the fraction-of-an-hour and retain accuracy. If the target resolution is
  816. // lower (i.e. > 1.0h) then we sum groups of hourly data by resolution to maintain fidelity.
  817. // e.g. (100 hours of per-hour hourly data, resolutionHours=10) => 10 data points, grouped and summed by 10-hour window
  818. // e.g. (20 minutes of per-minute hourly data, resolutionHours=1/60) => 20 data points, scaled down by a factor of 60
  819. func ScaleHourlyCostData(data map[string]*CostData, resolutionHours float64) map[string]*CostData {
  820. scaled := map[string]*CostData{}
  821. for key, datum := range data {
  822. datum.RAMReq = scaleVectorSeries(datum.RAMReq, resolutionHours)
  823. datum.RAMUsed = scaleVectorSeries(datum.RAMUsed, resolutionHours)
  824. datum.RAMAllocation = scaleVectorSeries(datum.RAMAllocation, resolutionHours)
  825. datum.CPUReq = scaleVectorSeries(datum.CPUReq, resolutionHours)
  826. datum.CPUUsed = scaleVectorSeries(datum.CPUUsed, resolutionHours)
  827. datum.CPUAllocation = scaleVectorSeries(datum.CPUAllocation, resolutionHours)
  828. datum.GPUReq = scaleVectorSeries(datum.GPUReq, resolutionHours)
  829. datum.NetworkData = scaleVectorSeries(datum.NetworkData, resolutionHours)
  830. for _, pvcDatum := range datum.PVCData {
  831. pvcDatum.Values = scaleVectorSeries(pvcDatum.Values, resolutionHours)
  832. }
  833. scaled[key] = datum
  834. }
  835. return scaled
  836. }
  837. func scaleVectorSeries(vs []*util.Vector, resolutionHours float64) []*util.Vector {
  838. // if scaling to a lower resolution, compress the hourly data for maximum accuracy
  839. if resolutionHours > 1.0 {
  840. return compressVectorSeries(vs, resolutionHours)
  841. }
  842. // if scaling to a higher resolution, simply scale each value down by the fraction of an hour
  843. for _, v := range vs {
  844. v.Value *= resolutionHours
  845. }
  846. return vs
  847. }
  848. func compressVectorSeries(vs []*util.Vector, resolutionHours float64) []*util.Vector {
  849. if len(vs) == 0 {
  850. return vs
  851. }
  852. compressed := []*util.Vector{}
  853. threshold := float64(60 * 60 * resolutionHours)
  854. var acc *util.Vector
  855. for i, v := range vs {
  856. if acc == nil {
  857. // start a new accumulation from current datum
  858. acc = &util.Vector{
  859. Value: vs[i].Value,
  860. Timestamp: vs[i].Timestamp,
  861. }
  862. continue
  863. }
  864. if v.Timestamp-acc.Timestamp < threshold {
  865. // v should be accumulated in current datum
  866. acc.Value += v.Value
  867. } else {
  868. // v falls outside current datum's threshold; append and start a new one
  869. compressed = append(compressed, acc)
  870. acc = &util.Vector{
  871. Value: vs[i].Value,
  872. Timestamp: vs[i].Timestamp,
  873. }
  874. }
  875. }
  876. // append any remaining, incomplete accumulation
  877. if acc != nil {
  878. compressed = append(compressed, acc)
  879. }
  880. return compressed
  881. }
  882. // ComputeAggregateCostModel computes cost data for the given window, then aggregates it by the given fields.
  883. // Data is cached on two levels: the aggregation is cached as well as the underlying cost data.
  884. func (a *Accesses) ComputeAggregateCostModel(promClient prometheusClient.Client, duration, offset, field string, subfields []string, rate string, filters map[string]string, sri *SharedResourceInfo, shared string, allocateIdle, includeTimeSeries, includeEfficiency, disableCache, clearCache, noCache, noExpireCache, remoteEnabled, disableSharedOverhead, useETLAdapter bool) (map[string]*Aggregation, string, error) {
  885. profileBaseName := fmt.Sprintf("ComputeAggregateCostModel(duration=%s, offet=%s, field=%s)", duration, offset, field)
  886. defer measureTime(time.Now(), profileThreshold, profileBaseName)
  887. // parse cost data filters into FilterFuncs
  888. filterFuncs := []FilterFunc{}
  889. retainFuncs := []FilterFunc{}
  890. retainFuncs = append(retainFuncs, func(cd *CostData) (bool, string) {
  891. if sri != nil {
  892. return sri.IsSharedResource(cd), ""
  893. }
  894. return false, ""
  895. })
  896. aggregateEnvironment := func(costDatum *CostData) string {
  897. if field == "cluster" {
  898. return costDatum.ClusterID
  899. } else if field == "node" {
  900. return costDatum.NodeName
  901. } else if field == "namespace" {
  902. return costDatum.Namespace
  903. } else if field == "service" {
  904. if len(costDatum.Services) > 0 {
  905. return costDatum.Namespace + "/" + costDatum.Services[0]
  906. }
  907. } else if field == "deployment" {
  908. if len(costDatum.Deployments) > 0 {
  909. return costDatum.Namespace + "/" + costDatum.Deployments[0]
  910. }
  911. } else if field == "daemonset" {
  912. if len(costDatum.Daemonsets) > 0 {
  913. return costDatum.Namespace + "/" + costDatum.Daemonsets[0]
  914. }
  915. } else if field == "statefulset" {
  916. if len(costDatum.Statefulsets) > 0 {
  917. return costDatum.Namespace + "/" + costDatum.Statefulsets[0]
  918. }
  919. } else if field == "label" {
  920. if costDatum.Labels != nil {
  921. for _, sf := range subfields {
  922. if subfieldName, ok := costDatum.Labels[sf]; ok {
  923. return fmt.Sprintf("%s=%s", sf, subfieldName)
  924. }
  925. }
  926. }
  927. } else if field == "pod" {
  928. return costDatum.Namespace + "/" + costDatum.PodName
  929. } else if field == "container" {
  930. return costDatum.Namespace + "/" + costDatum.PodName + "/" + costDatum.Name
  931. }
  932. return ""
  933. }
  934. if filters["podprefix"] != "" {
  935. pps := []string{}
  936. for _, fp := range strings.Split(filters["podprefix"], ",") {
  937. if fp != "" {
  938. cleanedFilter := strings.TrimSpace(fp)
  939. pps = append(pps, cleanedFilter)
  940. }
  941. }
  942. filterFuncs = append(filterFuncs, func(cd *CostData) (bool, string) {
  943. aggEnv := aggregateEnvironment(cd)
  944. for _, pp := range pps {
  945. cleanedFilter := strings.TrimSpace(pp)
  946. if strings.HasPrefix(cd.PodName, cleanedFilter) {
  947. return true, aggEnv
  948. }
  949. }
  950. return false, aggEnv
  951. })
  952. }
  953. if filters["namespace"] != "" {
  954. // namespaces may be comma-separated, e.g. kubecost,default
  955. // multiple namespaces are evaluated as an OR relationship
  956. nss := strings.Split(filters["namespace"], ",")
  957. filterFuncs = append(filterFuncs, func(cd *CostData) (bool, string) {
  958. aggEnv := aggregateEnvironment(cd)
  959. for _, ns := range nss {
  960. nsTrim := strings.TrimSpace(ns)
  961. if cd.Namespace == nsTrim {
  962. return true, aggEnv
  963. } else if strings.HasSuffix(nsTrim, "*") { // trigger wildcard prefix filtering
  964. nsTrimAsterisk := strings.TrimSuffix(nsTrim, "*")
  965. if strings.HasPrefix(cd.Namespace, nsTrimAsterisk) {
  966. return true, aggEnv
  967. }
  968. }
  969. }
  970. return false, aggEnv
  971. })
  972. }
  973. if filters["node"] != "" {
  974. // nodes may be comma-separated, e.g. aws-node-1,aws-node-2
  975. // multiple nodes are evaluated as an OR relationship
  976. nodes := strings.Split(filters["node"], ",")
  977. filterFuncs = append(filterFuncs, func(cd *CostData) (bool, string) {
  978. aggEnv := aggregateEnvironment(cd)
  979. for _, node := range nodes {
  980. nodeTrim := strings.TrimSpace(node)
  981. if cd.NodeName == nodeTrim {
  982. return true, aggEnv
  983. } else if strings.HasSuffix(nodeTrim, "*") { // trigger wildcard prefix filtering
  984. nodeTrimAsterisk := strings.TrimSuffix(nodeTrim, "*")
  985. if strings.HasPrefix(cd.NodeName, nodeTrimAsterisk) {
  986. return true, aggEnv
  987. }
  988. }
  989. }
  990. return false, aggEnv
  991. })
  992. }
  993. if filters["cluster"] != "" {
  994. // clusters may be comma-separated, e.g. cluster-one,cluster-two
  995. // multiple clusters are evaluated as an OR relationship
  996. cs := strings.Split(filters["cluster"], ",")
  997. filterFuncs = append(filterFuncs, func(cd *CostData) (bool, string) {
  998. aggEnv := aggregateEnvironment(cd)
  999. for _, c := range cs {
  1000. cTrim := strings.TrimSpace(c)
  1001. id, name := cd.ClusterID, cd.ClusterName
  1002. if id == cTrim || name == cTrim {
  1003. return true, aggEnv
  1004. } else if strings.HasSuffix(cTrim, "*") { // trigger wildcard prefix filtering
  1005. cTrimAsterisk := strings.TrimSuffix(cTrim, "*")
  1006. if strings.HasPrefix(id, cTrimAsterisk) || strings.HasPrefix(name, cTrimAsterisk) {
  1007. return true, aggEnv
  1008. }
  1009. }
  1010. }
  1011. return false, aggEnv
  1012. })
  1013. }
  1014. if filters["labels"] != "" {
  1015. // labels are expected to be comma-separated and to take the form key=value
  1016. // e.g. app=cost-analyzer,app.kubernetes.io/instance=kubecost
  1017. // each different label will be applied as an AND
  1018. // multiple values for a single label will be evaluated as an OR
  1019. labelValues := map[string][]string{}
  1020. ls := strings.Split(filters["labels"], ",")
  1021. for _, l := range ls {
  1022. lTrim := strings.TrimSpace(l)
  1023. label := strings.Split(lTrim, "=")
  1024. if len(label) == 2 {
  1025. ln := prom.SanitizeLabelName(strings.TrimSpace(label[0]))
  1026. lv := strings.TrimSpace(label[1])
  1027. labelValues[ln] = append(labelValues[ln], lv)
  1028. } else {
  1029. // label is not of the form name=value, so log it and move on
  1030. klog.V(2).Infof("[Warning] aggregate cost model: skipping illegal label filter: %s", l)
  1031. }
  1032. }
  1033. // Generate FilterFunc for each set of label filters by invoking a function instead of accessing
  1034. // values by closure to prevent reference-type looping bug.
  1035. // (see https://github.com/golang/go/wiki/CommonMistakes#using-reference-to-loop-iterator-variable)
  1036. for label, values := range labelValues {
  1037. ff := (func(l string, vs []string) FilterFunc {
  1038. return func(cd *CostData) (bool, string) {
  1039. ae := aggregateEnvironment(cd)
  1040. for _, v := range vs {
  1041. if v == "__unallocated__" { // Special case. __unallocated__ means return all pods without the attached label
  1042. if _, ok := cd.Labels[label]; !ok {
  1043. return true, ae
  1044. }
  1045. }
  1046. if cd.Labels[label] == v {
  1047. return true, ae
  1048. } else if strings.HasSuffix(v, "*") { // trigger wildcard prefix filtering
  1049. vTrim := strings.TrimSuffix(v, "*")
  1050. if strings.HasPrefix(cd.Labels[label], vTrim) {
  1051. return true, ae
  1052. }
  1053. }
  1054. }
  1055. return false, ae
  1056. }
  1057. })(label, values)
  1058. filterFuncs = append(filterFuncs, ff)
  1059. }
  1060. }
  1061. // clear cache prior to checking the cache so that a clearCache=true
  1062. // request always returns a freshly computed value
  1063. if clearCache {
  1064. a.AggregateCache.Flush()
  1065. a.CostDataCache.Flush()
  1066. }
  1067. cacheExpiry := a.GetCacheExpiration(duration)
  1068. if noExpireCache {
  1069. cacheExpiry = cache.NoExpiration
  1070. }
  1071. // parametrize cache key by all request parameters
  1072. aggKey := GenerateAggKey(aggKeyParams{
  1073. duration: duration,
  1074. offset: offset,
  1075. filters: filters,
  1076. field: field,
  1077. subfields: subfields,
  1078. rate: rate,
  1079. sri: sri,
  1080. shareType: shared,
  1081. idle: allocateIdle,
  1082. timeSeries: includeTimeSeries,
  1083. efficiency: includeEfficiency,
  1084. })
  1085. // convert duration and offset to start and end times
  1086. startTime, endTime, err := ParseTimeRange(duration, offset)
  1087. if err != nil {
  1088. return nil, "", fmt.Errorf("Error parsing duration (%s) and offset (%s): %s", duration, offset, err)
  1089. }
  1090. durationHours := endTime.Sub(*startTime).Hours()
  1091. thanosOffset := time.Now().Add(-thanos.OffsetDuration())
  1092. if a.ThanosClient != nil && endTime.After(thanosOffset) {
  1093. klog.V(4).Infof("Setting end time backwards to first present data")
  1094. // Apply offsets to both end and start times to maintain correct time range
  1095. deltaDuration := endTime.Sub(thanosOffset)
  1096. *startTime = startTime.Add(-1 * deltaDuration)
  1097. *endTime = time.Now().Add(-thanos.OffsetDuration())
  1098. }
  1099. // determine resolution by size of duration
  1100. resolutionHours := durationHours
  1101. if durationHours >= 2160 {
  1102. // 90 days
  1103. resolutionHours = 72.0
  1104. } else if durationHours >= 720 {
  1105. // 30 days
  1106. resolutionHours = 24.0
  1107. } else if durationHours >= 168 {
  1108. // 7 days
  1109. resolutionHours = 24.0
  1110. } else if durationHours >= 48 {
  1111. // 2 days
  1112. resolutionHours = 2.0
  1113. } else if durationHours >= 1 {
  1114. resolutionHours = 1.0
  1115. }
  1116. key := fmt.Sprintf(`%s:%s:%fh:%t`, duration, offset, resolutionHours, remoteEnabled)
  1117. // report message about which of the two caches hit. by default report a miss
  1118. cacheMessage := fmt.Sprintf("L1 cache miss: %s L2 cache miss: %s", aggKey, key)
  1119. // check the cache for aggregated response; if cache is hit and not disabled, return response
  1120. if value, found := a.AggregateCache.Get(aggKey); found && !disableCache && !noCache {
  1121. result, ok := value.(map[string]*Aggregation)
  1122. if !ok {
  1123. // disable cache and recompute if type cast fails
  1124. klog.Errorf("caching error: failed to cast aggregate data to struct: %s", aggKey)
  1125. return a.ComputeAggregateCostModel(promClient, duration, offset, field, subfields, rate, filters, sri, shared, allocateIdle, includeTimeSeries, includeEfficiency, true, false, noExpireCache, noCache, remoteEnabled, disableSharedOverhead, useETLAdapter)
  1126. }
  1127. return result, fmt.Sprintf("aggregate cache hit: %s", aggKey), nil
  1128. }
  1129. profileStart := time.Now()
  1130. profileName := profileBaseName + ": "
  1131. window := duration
  1132. if durationHours >= 1 {
  1133. window = fmt.Sprintf("%dh", int(resolutionHours))
  1134. // exclude the last window of the time frame to match Prometheus definitions of range, offset, and resolution
  1135. *startTime = startTime.Add(time.Duration(resolutionHours) * time.Hour)
  1136. } else {
  1137. // don't cache requests for durations of less than one hour
  1138. klog.Infof("key %s has durationhours %f", key, durationHours)
  1139. disableCache = true
  1140. }
  1141. profileBaseName = fmt.Sprintf("ComputeAggregateCostModel(duration=%s, offset=%s, field=%s, window=%s)", duration, offset, field, window)
  1142. // attempt to retrieve cost data from cache
  1143. var costData map[string]*CostData
  1144. cacheData, found := a.CostDataCache.Get(key)
  1145. if found && !disableCache && !noCache {
  1146. profileName += "get cost data from cache"
  1147. ok := false
  1148. costData, ok = cacheData.(map[string]*CostData)
  1149. cacheMessage = fmt.Sprintf("L1 cache miss: %s, L2 cost data cache hit: %s", aggKey, key)
  1150. if !ok {
  1151. klog.Errorf("caching error: failed to cast cost data to struct: %s", key)
  1152. }
  1153. } else {
  1154. klog.Infof("key %s missed cache. found %t, disableCache %t, noCache %t ", key, found, disableCache, noCache)
  1155. cv := a.CostDataCache.Items()
  1156. klog.V(3).Infof("Logging cache items...")
  1157. for k := range cv {
  1158. klog.V(3).Infof("Cache item: %s", k)
  1159. }
  1160. profileName += "compute cost data"
  1161. start := startTime.Format(RFC3339Milli)
  1162. end := endTime.Format(RFC3339Milli)
  1163. costData, err = a.Model.ComputeCostDataRange(promClient, a.CloudProvider, start, end, window, resolutionHours, "", "", remoteEnabled, offset)
  1164. if err != nil {
  1165. if prom.IsErrorCollection(err) {
  1166. return nil, "", err
  1167. }
  1168. if pce, ok := err.(prom.CommError); ok {
  1169. return nil, "", pce
  1170. }
  1171. if strings.Contains(err.Error(), "data is empty") {
  1172. return nil, "", &EmptyDataError{err: err, duration: duration, offset: offset}
  1173. }
  1174. return nil, "", err
  1175. }
  1176. // compute length of the time series in the cost data and only compute
  1177. // aggregates and cache if the length is sufficiently high
  1178. costDataLen := costDataTimeSeriesLength(costData)
  1179. if durationHours < 1.0 {
  1180. // scale hourly cost data down to fractional hour
  1181. costData = ScaleHourlyCostData(costData, resolutionHours)
  1182. }
  1183. if costDataLen == 0 {
  1184. return nil, "", &EmptyDataError{duration: duration, offset: offset}
  1185. }
  1186. if costDataLen >= minCostDataLength && !noCache {
  1187. klog.Infof("Setting L2 cache: %s", key)
  1188. a.CostDataCache.Set(key, costData, cacheExpiry)
  1189. }
  1190. }
  1191. measureTime(profileStart, profileThreshold, profileName)
  1192. c, err := a.CloudProvider.GetConfig()
  1193. if err != nil {
  1194. return nil, "", err
  1195. }
  1196. discount, err := ParsePercentString(c.Discount)
  1197. if err != nil {
  1198. return nil, "", err
  1199. }
  1200. customDiscount, err := ParsePercentString(c.NegotiatedDiscount)
  1201. if err != nil {
  1202. return nil, "", err
  1203. }
  1204. sc := make(map[string]*SharedCostInfo)
  1205. if !disableSharedOverhead {
  1206. for key, val := range c.SharedCosts {
  1207. cost, err := strconv.ParseFloat(val, 64)
  1208. durationCoefficient := durationHours / util.HoursPerMonth
  1209. if err != nil {
  1210. return nil, "", fmt.Errorf("Unable to parse shared cost %s: %s", val, err.Error())
  1211. }
  1212. sc[key] = &SharedCostInfo{
  1213. Name: key,
  1214. Cost: cost * durationCoefficient,
  1215. }
  1216. }
  1217. }
  1218. profileStart = time.Now()
  1219. profileName = profileBaseName + ": compute idle coefficient"
  1220. idleCoefficients := make(map[string]float64)
  1221. if allocateIdle {
  1222. idleDurationCalcHours := durationHours
  1223. if durationHours < 1 {
  1224. idleDurationCalcHours = 1
  1225. }
  1226. windowStr := fmt.Sprintf("%dh", int(idleDurationCalcHours))
  1227. if a.ThanosClient != nil {
  1228. offset = thanos.Offset()
  1229. klog.Infof("Setting offset to %s", offset)
  1230. }
  1231. idleCoefficients, err = a.ComputeIdleCoefficient(costData, promClient, a.CloudProvider, discount, customDiscount, windowStr, offset)
  1232. if err != nil {
  1233. klog.Errorf("error computing idle coefficient: windowString=%s, offset=%s, err=%s", windowStr, offset, err)
  1234. return nil, "", err
  1235. }
  1236. }
  1237. for cid, idleCoefficient := range idleCoefficients {
  1238. klog.Infof("Idle Coeff: %s: %f", cid, idleCoefficient)
  1239. }
  1240. totalContainerCost := 0.0
  1241. if shared == SplitTypeWeighted {
  1242. totalContainerCost = GetTotalContainerCost(costData, rate, a.CloudProvider, discount, customDiscount, idleCoefficients)
  1243. }
  1244. measureTime(profileStart, profileThreshold, profileName)
  1245. profileStart = time.Now()
  1246. profileName = profileBaseName + ": filter cost data"
  1247. // filter cost data by namespace and cluster after caching for maximal cache hits
  1248. costData, filteredContainerCount, filteredEnvironments := FilterCostData(costData, retainFuncs, filterFuncs)
  1249. measureTime(profileStart, profileThreshold, profileName)
  1250. profileStart = time.Now()
  1251. profileName = profileBaseName + ": aggregate cost data"
  1252. // aggregate cost model data by given fields and cache the result for the default expiration
  1253. opts := &AggregationOptions{
  1254. Discount: discount,
  1255. CustomDiscount: customDiscount,
  1256. IdleCoefficients: idleCoefficients,
  1257. IncludeEfficiency: includeEfficiency,
  1258. IncludeTimeSeries: includeTimeSeries,
  1259. Rate: rate,
  1260. ResolutionHours: resolutionHours,
  1261. SharedResourceInfo: sri,
  1262. SharedCosts: sc,
  1263. FilteredContainerCount: filteredContainerCount,
  1264. FilteredEnvironments: filteredEnvironments,
  1265. TotalContainerCost: totalContainerCost,
  1266. SharedSplit: shared,
  1267. }
  1268. result := AggregateCostData(costData, field, subfields, a.CloudProvider, opts)
  1269. // If sending time series data back, switch scale back to hourly data. At this point,
  1270. // resolutionHours may have converted our hourly data to more- or less-than hourly data.
  1271. if includeTimeSeries {
  1272. for _, aggs := range result {
  1273. ScaleAggregationTimeSeries(aggs, resolutionHours)
  1274. }
  1275. }
  1276. // compute length of the time series in the cost data and only cache
  1277. // aggregation results if the length is sufficiently high
  1278. costDataLen := costDataTimeSeriesLength(costData)
  1279. if costDataLen >= minCostDataLength && durationHours > 1 && !noCache {
  1280. // Set the result map (rather than a pointer to it) because map is a reference type
  1281. klog.Infof("Caching key in aggregate cache: %s", key)
  1282. a.AggregateCache.Set(aggKey, result, cacheExpiry)
  1283. } else {
  1284. klog.Infof("Not caching for key %s. Not enough data: %t, Duration less than 1h: %t, noCache: %t", key, costDataLen < minCostDataLength, durationHours < 1, noCache)
  1285. }
  1286. measureTime(profileStart, profileThreshold, profileName)
  1287. return result, cacheMessage, nil
  1288. }
  1289. // ScaleAggregationTimeSeries reverses the scaling done by ScaleHourlyCostData, returning
  1290. // the aggregation's time series to hourly data.
  1291. func ScaleAggregationTimeSeries(aggregation *Aggregation, resolutionHours float64) {
  1292. for _, v := range aggregation.CPUCostVector {
  1293. v.Value /= resolutionHours
  1294. }
  1295. for _, v := range aggregation.GPUCostVector {
  1296. v.Value /= resolutionHours
  1297. }
  1298. for _, v := range aggregation.RAMCostVector {
  1299. v.Value /= resolutionHours
  1300. }
  1301. for _, v := range aggregation.PVCostVector {
  1302. v.Value /= resolutionHours
  1303. }
  1304. for _, v := range aggregation.NetworkCostVector {
  1305. v.Value /= resolutionHours
  1306. }
  1307. for _, v := range aggregation.TotalCostVector {
  1308. v.Value /= resolutionHours
  1309. }
  1310. return
  1311. }
  1312. // String returns a string representation of the encapsulated shared resources, which
  1313. // can be used to uniquely identify a set of shared resources. Sorting sets of shared
  1314. // resources ensures that strings representing permutations of the same combination match.
  1315. func (s *SharedResourceInfo) String() string {
  1316. if s == nil {
  1317. return ""
  1318. }
  1319. nss := []string{}
  1320. for ns := range s.SharedNamespace {
  1321. nss = append(nss, ns)
  1322. }
  1323. sort.Strings(nss)
  1324. nsStr := strings.Join(nss, ",")
  1325. labels := []string{}
  1326. for lbl, vals := range s.LabelSelectors {
  1327. for val := range vals {
  1328. if lbl != "" && val != "" {
  1329. labels = append(labels, fmt.Sprintf("%s=%s", lbl, val))
  1330. }
  1331. }
  1332. }
  1333. sort.Strings(labels)
  1334. labelStr := strings.Join(labels, ",")
  1335. return fmt.Sprintf("%s:%s", nsStr, labelStr)
  1336. }
  1337. type aggKeyParams struct {
  1338. duration string
  1339. offset string
  1340. filters map[string]string
  1341. field string
  1342. subfields []string
  1343. rate string
  1344. sri *SharedResourceInfo
  1345. shareType string
  1346. idle bool
  1347. timeSeries bool
  1348. efficiency bool
  1349. }
  1350. // GenerateAggKey generates a parameter-unique key for caching the aggregate cost model
  1351. func GenerateAggKey(ps aggKeyParams) string {
  1352. // parse, trim, and sort podprefix filters
  1353. podPrefixFilters := []string{}
  1354. if ppfs, ok := ps.filters["podprefix"]; ok && ppfs != "" {
  1355. for _, psf := range strings.Split(ppfs, ",") {
  1356. podPrefixFilters = append(podPrefixFilters, strings.TrimSpace(psf))
  1357. }
  1358. }
  1359. sort.Strings(podPrefixFilters)
  1360. podPrefixFiltersStr := strings.Join(podPrefixFilters, ",")
  1361. // parse, trim, and sort namespace filters
  1362. nsFilters := []string{}
  1363. if nsfs, ok := ps.filters["namespace"]; ok && nsfs != "" {
  1364. for _, nsf := range strings.Split(nsfs, ",") {
  1365. nsFilters = append(nsFilters, strings.TrimSpace(nsf))
  1366. }
  1367. }
  1368. sort.Strings(nsFilters)
  1369. nsFilterStr := strings.Join(nsFilters, ",")
  1370. // parse, trim, and sort node filters
  1371. nodeFilters := []string{}
  1372. if nodefs, ok := ps.filters["node"]; ok && nodefs != "" {
  1373. for _, nodef := range strings.Split(nodefs, ",") {
  1374. nodeFilters = append(nodeFilters, strings.TrimSpace(nodef))
  1375. }
  1376. }
  1377. sort.Strings(nodeFilters)
  1378. nodeFilterStr := strings.Join(nodeFilters, ",")
  1379. // parse, trim, and sort cluster filters
  1380. cFilters := []string{}
  1381. if cfs, ok := ps.filters["cluster"]; ok && cfs != "" {
  1382. for _, cf := range strings.Split(cfs, ",") {
  1383. cFilters = append(cFilters, strings.TrimSpace(cf))
  1384. }
  1385. }
  1386. sort.Strings(cFilters)
  1387. cFilterStr := strings.Join(cFilters, ",")
  1388. // parse, trim, and sort label filters
  1389. lFilters := []string{}
  1390. if lfs, ok := ps.filters["labels"]; ok && lfs != "" {
  1391. for _, lf := range strings.Split(lfs, ",") {
  1392. // trim whitespace from the label name and the label value
  1393. // of each label name/value pair, then reconstruct
  1394. // e.g. "tier = frontend, app = kubecost" == "app=kubecost,tier=frontend"
  1395. lfa := strings.Split(lf, "=")
  1396. if len(lfa) == 2 {
  1397. lfn := strings.TrimSpace(lfa[0])
  1398. lfv := strings.TrimSpace(lfa[1])
  1399. lFilters = append(lFilters, fmt.Sprintf("%s=%s", lfn, lfv))
  1400. } else {
  1401. // label is not of the form name=value, so log it and move on
  1402. klog.V(2).Infof("[Warning] GenerateAggKey: skipping illegal label filter: %s", lf)
  1403. }
  1404. }
  1405. }
  1406. sort.Strings(lFilters)
  1407. lFilterStr := strings.Join(lFilters, ",")
  1408. filterStr := fmt.Sprintf("%s:%s:%s:%s:%s", nsFilterStr, nodeFilterStr, cFilterStr, lFilterStr, podPrefixFiltersStr)
  1409. sort.Strings(ps.subfields)
  1410. fieldStr := fmt.Sprintf("%s:%s", ps.field, strings.Join(ps.subfields, ","))
  1411. return fmt.Sprintf("%s:%s:%s:%s:%s:%s:%s:%t:%t:%t", ps.duration, ps.offset, filterStr, fieldStr, ps.rate,
  1412. ps.sri, ps.shareType, ps.idle, ps.timeSeries, ps.efficiency)
  1413. }
  1414. // Aggregator is capable of computing the aggregated cost model. This is
  1415. // a brutal interface, which should be cleaned up, but it's necessary for
  1416. // being able to swap in an ETL-backed implementation.
  1417. // TODO clean up, simplify
  1418. type Aggregator interface {
  1419. ComputeAggregateCostModel(promClient prometheusClient.Client, duration, offset, field string, subfields []string, rate string, filters map[string]string, sri *SharedResourceInfo, shared string, allocateIdle, includeTimeSeries, includeEfficiency, disableCache, clearCache, noCache, noExpireCache, remoteEnabled, disableSharedOverhead, useETLAdapter bool) (map[string]*Aggregation, string, error)
  1420. }
  1421. // AggregateCostModelHandler handles requests to the aggregated cost model API. See
  1422. // ComputeAggregateCostModel for details.
  1423. func (a *Accesses) AggregateCostModelHandler(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  1424. w.Header().Set("Content-Type", "application/json")
  1425. windowStr := r.URL.Query().Get("window")
  1426. // Convert UTC-RFC3339 pairs to configured UTC offset
  1427. // e.g. with UTC offset of -0600, 2020-07-01T00:00:00Z becomes
  1428. // 2020-07-01T06:00:00Z == 2020-07-01T00:00:00-0600
  1429. // TODO niko/etl fix the frontend because this is confusing if you're
  1430. // actually asking for UTC time (...Z) and we swap that "Z" out for the
  1431. // configured UTC offset without asking
  1432. rfc3339 := `\d\d\d\d-\d\d-\d\dT\d\d:\d\d:\d\dZ`
  1433. regex := regexp.MustCompile(fmt.Sprintf(`(%s),(%s)`, rfc3339, rfc3339))
  1434. match := regex.FindStringSubmatch(windowStr)
  1435. if match != nil {
  1436. start, _ := time.Parse(time.RFC3339, match[1])
  1437. start = start.Add(-env.GetParsedUTCOffset()).In(time.UTC)
  1438. end, _ := time.Parse(time.RFC3339, match[2])
  1439. end = end.Add(-env.GetParsedUTCOffset()).In(time.UTC)
  1440. windowStr = fmt.Sprintf("%sZ,%sZ", start.Format("2006-01-02T15:04:05"), end.Format("2006-01-02T15:04:05Z"))
  1441. }
  1442. // determine duration and offset from query parameters
  1443. window, err := kubecost.ParseWindowWithOffset(windowStr, env.GetParsedUTCOffset())
  1444. if err != nil || window.Start() == nil {
  1445. http.Error(w, fmt.Sprintf("invalid window: %s", err), http.StatusBadRequest)
  1446. return
  1447. }
  1448. duration, offset := window.ToDurationOffset()
  1449. durRegex := regexp.MustCompile(`^(\d+)(m|h|d|s)$`)
  1450. isDurationStr := durRegex.MatchString(windowStr)
  1451. // legacy offset option should override window offset
  1452. if r.URL.Query().Get("offset") != "" {
  1453. offset = r.URL.Query().Get("offset")
  1454. // Shift window by offset, but only when manually set with separate
  1455. // parameter and window was provided as a duration string. Otherwise,
  1456. // do not alter the (duration, offset) from ParseWindowWithOffset.
  1457. if offset != "1m" && isDurationStr {
  1458. match := durRegex.FindStringSubmatch(offset)
  1459. if match != nil && len(match) == 3 {
  1460. dur := time.Minute
  1461. if match[2] == "h" {
  1462. dur = time.Hour
  1463. }
  1464. if match[2] == "d" {
  1465. dur = 24 * time.Hour
  1466. }
  1467. if match[2] == "s" {
  1468. dur = time.Second
  1469. }
  1470. num, _ := strconv.ParseInt(match[1], 10, 64)
  1471. window = window.Shift(-time.Duration(num) * dur)
  1472. }
  1473. }
  1474. }
  1475. // redirect requests with no offset to a 1m offset to improve cache hits
  1476. if offset == "" {
  1477. offset = "1m"
  1478. }
  1479. // parse remaining query parameters
  1480. namespace := r.URL.Query().Get("namespace")
  1481. cluster := r.URL.Query().Get("cluster")
  1482. labels := r.URL.Query().Get("labels")
  1483. podprefix := r.URL.Query().Get("podprefix")
  1484. labelArray := strings.Split(labels, "=")
  1485. labelArray[0] = strings.ReplaceAll(labelArray[0], "-", "_")
  1486. labels = strings.Join(labelArray, "=")
  1487. field := r.URL.Query().Get("aggregation")
  1488. subfieldStr := r.URL.Query().Get("aggregationSubfield")
  1489. rate := r.URL.Query().Get("rate")
  1490. idleFlag := r.URL.Query().Get("allocateIdle")
  1491. sharedNamespaces := r.URL.Query().Get("sharedNamespaces")
  1492. sharedLabelNames := r.URL.Query().Get("sharedLabelNames")
  1493. sharedLabelValues := r.URL.Query().Get("sharedLabelValues")
  1494. remote := r.URL.Query().Get("remote") != "false"
  1495. shared := r.URL.Query().Get("sharedSplit")
  1496. subfields := []string{}
  1497. if len(subfieldStr) > 0 {
  1498. s := strings.Split(r.URL.Query().Get("aggregationSubfield"), ",")
  1499. for _, rawLabel := range s {
  1500. subfields = append(subfields, prom.SanitizeLabelName(rawLabel))
  1501. }
  1502. }
  1503. var allocateIdle bool
  1504. if idleFlag == "default" {
  1505. c, _ := a.CloudProvider.GetConfig()
  1506. allocateIdle = (c.DefaultIdle == "true")
  1507. } else {
  1508. allocateIdle = (idleFlag == "true")
  1509. }
  1510. // timeSeries == true maintains the time series dimension of the data,
  1511. // which by default gets summed over the entire interval
  1512. includeTimeSeries := r.URL.Query().Get("timeSeries") == "true"
  1513. // efficiency == true aggregates and returns usage and efficiency data
  1514. // includeEfficiency := r.URL.Query().Get("efficiency") == "true"
  1515. // efficiency has been deprecated in favor of a default to always send efficiency
  1516. includeEfficiency := true
  1517. // TODO niko/caching rename "recomputeCache"
  1518. // disableCache, if set to "true", tells this function to recompute and
  1519. // cache the requested data
  1520. disableCache := r.URL.Query().Get("disableCache") == "true"
  1521. // clearCache, if set to "true", tells this function to flush the cache,
  1522. // then recompute and cache the requested data
  1523. clearCache := r.URL.Query().Get("clearCache") == "true"
  1524. // noCache avoids the cache altogether, both reading from and writing to
  1525. noCache := r.URL.Query().Get("noCache") == "true"
  1526. // noExpireCache should only be used by cache warming to set non-expiring caches
  1527. noExpireCache := false
  1528. // aggregation field is required
  1529. if field == "" {
  1530. http.Error(w, "Missing aggregation field parameter", http.StatusBadRequest)
  1531. return
  1532. }
  1533. // aggregation subfield is required when aggregation field is "label"
  1534. if field == "label" && len(subfields) == 0 {
  1535. http.Error(w, "Missing aggregation subfield parameter for aggregation by label", http.StatusBadRequest)
  1536. return
  1537. }
  1538. // enforce one of four available rate options
  1539. if rate != "" && rate != "hourly" && rate != "daily" && rate != "monthly" {
  1540. http.Error(w, "If set, rate parameter must be one of: 'hourly', 'daily', 'monthly'", http.StatusBadRequest)
  1541. return
  1542. }
  1543. // parse cost data filters
  1544. // namespace and cluster are exact-string-matches
  1545. // labels are expected to be comma-separated and to take the form key=value
  1546. // e.g. app=cost-analyzer,app.kubernetes.io/instance=kubecost
  1547. filters := map[string]string{
  1548. "namespace": namespace,
  1549. "cluster": cluster,
  1550. "labels": labels,
  1551. "podprefix": podprefix,
  1552. }
  1553. // parse shared resources
  1554. sn := []string{}
  1555. sln := []string{}
  1556. slv := []string{}
  1557. if sharedNamespaces != "" {
  1558. sn = strings.Split(sharedNamespaces, ",")
  1559. }
  1560. if sharedLabelNames != "" {
  1561. sln = strings.Split(sharedLabelNames, ",")
  1562. slv = strings.Split(sharedLabelValues, ",")
  1563. if len(sln) != len(slv) || slv[0] == "" {
  1564. http.Error(w, "Supply exacly one shared label value per shared label name", http.StatusBadRequest)
  1565. return
  1566. }
  1567. }
  1568. var sr *SharedResourceInfo
  1569. if len(sn) > 0 || len(sln) > 0 {
  1570. sr = NewSharedResourceInfo(true, sn, sln, slv)
  1571. }
  1572. // enable remote if it is available and not disabled
  1573. remoteEnabled := remote && env.IsRemoteEnabled()
  1574. promClient := a.GetPrometheusClient(remote)
  1575. useETLAdapter := r.URL.Query().Get("etl") == "true"
  1576. var data map[string]*Aggregation
  1577. var message string
  1578. data, message, err = a.AggAPI.ComputeAggregateCostModel(promClient, duration, offset, field, subfields, rate, filters, sr, shared, allocateIdle, includeTimeSeries, includeEfficiency, disableCache, clearCache, noCache, noExpireCache, remoteEnabled, false, useETLAdapter)
  1579. // Find any warnings in http request context
  1580. warning, _ := util.GetWarning(r)
  1581. if err != nil {
  1582. if emptyErr, ok := err.(*EmptyDataError); ok {
  1583. if warning == "" {
  1584. w.Write(WrapData(map[string]interface{}{}, emptyErr))
  1585. } else {
  1586. w.Write(WrapDataWithWarning(map[string]interface{}{}, emptyErr, warning))
  1587. }
  1588. return
  1589. }
  1590. if boundaryErr, ok := err.(*kubecost.BoundaryError); ok {
  1591. if window.Start() != nil && window.Start().After(time.Now().Add(-90*24*time.Hour)) {
  1592. // Asking for data within a 90 day period: it will be available
  1593. // after the pipeline builds
  1594. msg := "Data will be available after ETL is built"
  1595. rex := regexp.MustCompile(`(\d+\.*\d*)%`)
  1596. match := rex.FindStringSubmatch(boundaryErr.Message)
  1597. if len(match) > 1 {
  1598. completionPct, err := strconv.ParseFloat(match[1], 64)
  1599. if err == nil {
  1600. msg = fmt.Sprintf("%s (%.1f%% complete)", msg, completionPct)
  1601. }
  1602. }
  1603. http.Error(w, msg, http.StatusInternalServerError)
  1604. } else {
  1605. // Boundary error outside of 90 day period; may not be available
  1606. http.Error(w, boundaryErr.Error(), http.StatusInternalServerError)
  1607. }
  1608. return
  1609. }
  1610. errStr := fmt.Sprintf("error computing aggregate cost model: %s", err)
  1611. http.Error(w, errStr, http.StatusInternalServerError)
  1612. return
  1613. }
  1614. if warning == "" {
  1615. w.Write(WrapDataWithMessage(data, nil, message))
  1616. } else {
  1617. w.Write(WrapDataWithMessageAndWarning(data, nil, message, warning))
  1618. }
  1619. }