costmodel.go 70 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129
  1. package costmodel
  2. import (
  3. "errors"
  4. "fmt"
  5. "maps"
  6. "math"
  7. "regexp"
  8. "strconv"
  9. "strings"
  10. "time"
  11. "github.com/opencost/opencost/core/pkg/clustercache"
  12. "github.com/opencost/opencost/core/pkg/clusters"
  13. coreenv "github.com/opencost/opencost/core/pkg/env"
  14. "github.com/opencost/opencost/core/pkg/filter/allocation"
  15. "github.com/opencost/opencost/core/pkg/log"
  16. "github.com/opencost/opencost/core/pkg/model/kubemodel"
  17. "github.com/opencost/opencost/core/pkg/opencost"
  18. "github.com/opencost/opencost/core/pkg/source"
  19. "github.com/opencost/opencost/core/pkg/util"
  20. "github.com/opencost/opencost/core/pkg/util/promutil"
  21. costAnalyzerCloud "github.com/opencost/opencost/pkg/cloud/models"
  22. km "github.com/opencost/opencost/pkg/kubemodel"
  23. v1 "k8s.io/api/core/v1"
  24. metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
  25. "k8s.io/apimachinery/pkg/labels"
  26. "golang.org/x/sync/singleflight"
  27. )
  28. const (
  29. profileThreshold = 1000 * 1000 * 1000 // 1s (in ns)
  30. unmountedPVsContainer = "unmounted-pvs"
  31. )
  32. // isCron matches a CronJob name and captures the non-timestamp name
  33. //
  34. // We support either a 10 character timestamp OR an 8 character timestamp
  35. // because batch/v1beta1 CronJobs creates Jobs with 10 character timestamps
  36. // and batch/v1 CronJobs create Jobs with 8 character timestamps.
  37. var isCron = regexp.MustCompile(`^(.+)-(\d{10}|\d{8})$`)
  38. type CostModel struct {
  39. Cache clustercache.ClusterCache
  40. ClusterMap clusters.ClusterMap
  41. BatchDuration time.Duration
  42. RequestGroup *singleflight.Group
  43. DataSource source.OpenCostDataSource
  44. Provider costAnalyzerCloud.Provider
  45. KubeModel *km.KubeModel
  46. pricingMetadata *costAnalyzerCloud.PricingMatchMetadata
  47. }
  48. func NewCostModel(
  49. clusterUID string,
  50. dataSource source.OpenCostDataSource,
  51. provider costAnalyzerCloud.Provider,
  52. cache clustercache.ClusterCache,
  53. clusterMap clusters.ClusterMap,
  54. batchDuration time.Duration,
  55. ) *CostModel {
  56. // request grouping to prevent over-requesting the same data prior to caching
  57. requestGroup := new(singleflight.Group)
  58. var kubeModel *km.KubeModel
  59. var err error
  60. if dataSource != nil {
  61. kubeModel, err = km.NewKubeModel(clusterUID, dataSource)
  62. if err != nil {
  63. // KubeModel is required. Log a fatal error if we fail to init.
  64. log.Fatalf("error initializing KubeModel: %s", err)
  65. }
  66. }
  67. return &CostModel{
  68. Cache: cache,
  69. ClusterMap: clusterMap,
  70. BatchDuration: batchDuration,
  71. DataSource: dataSource,
  72. Provider: provider,
  73. RequestGroup: requestGroup,
  74. KubeModel: kubeModel,
  75. }
  76. }
  77. func (cm *CostModel) ComputeKubeModelSet(start, end time.Time) (*kubemodel.KubeModelSet, error) {
  78. if cm.KubeModel == nil {
  79. return nil, fmt.Errorf("KubeModel not initialized")
  80. }
  81. return cm.KubeModel.ComputeKubeModelSet(start, end)
  82. }
  83. type CostData struct {
  84. Name string `json:"name,omitempty"`
  85. PodName string `json:"podName,omitempty"`
  86. NodeName string `json:"nodeName,omitempty"`
  87. NodeData *costAnalyzerCloud.Node `json:"node,omitempty"`
  88. Namespace string `json:"namespace,omitempty"`
  89. Deployments []string `json:"deployments,omitempty"`
  90. Services []string `json:"services,omitempty"`
  91. Daemonsets []string `json:"daemonsets,omitempty"`
  92. Statefulsets []string `json:"statefulsets,omitempty"`
  93. Jobs []string `json:"jobs,omitempty"`
  94. RAMReq []*util.Vector `json:"ramreq,omitempty"`
  95. RAMUsed []*util.Vector `json:"ramused,omitempty"`
  96. RAMAllocation []*util.Vector `json:"ramallocated,omitempty"`
  97. CPUReq []*util.Vector `json:"cpureq,omitempty"`
  98. CPUUsed []*util.Vector `json:"cpuused,omitempty"`
  99. CPUAllocation []*util.Vector `json:"cpuallocated,omitempty"`
  100. GPUReq []*util.Vector `json:"gpureq,omitempty"`
  101. PVCData []*PersistentVolumeClaimData `json:"pvcData,omitempty"`
  102. NetworkData []*util.Vector `json:"network,omitempty"`
  103. Annotations map[string]string `json:"annotations,omitempty"`
  104. Labels map[string]string `json:"labels,omitempty"`
  105. NamespaceLabels map[string]string `json:"namespaceLabels,omitempty"`
  106. ClusterID string `json:"clusterId"`
  107. ClusterName string `json:"clusterName"`
  108. }
  109. func (cd *CostData) String() string {
  110. return fmt.Sprintf("\n\tName: %s; PodName: %s, NodeName: %s\n\tNamespace: %s\n\tDeployments: %s\n\tServices: %s\n\tCPU (req, used, alloc): %d, %d, %d\n\tRAM (req, used, alloc): %d, %d, %d",
  111. cd.Name, cd.PodName, cd.NodeName, cd.Namespace, strings.Join(cd.Deployments, ", "), strings.Join(cd.Services, ", "),
  112. len(cd.CPUReq), len(cd.CPUUsed), len(cd.CPUAllocation),
  113. len(cd.RAMReq), len(cd.RAMUsed), len(cd.RAMAllocation))
  114. }
  115. func (cd *CostData) GetController() (name string, kind string, hasController bool) {
  116. hasController = false
  117. if len(cd.Deployments) > 0 {
  118. name = cd.Deployments[0]
  119. kind = "deployment"
  120. hasController = true
  121. } else if len(cd.Statefulsets) > 0 {
  122. name = cd.Statefulsets[0]
  123. kind = "statefulset"
  124. hasController = true
  125. } else if len(cd.Daemonsets) > 0 {
  126. name = cd.Daemonsets[0]
  127. kind = "daemonset"
  128. hasController = true
  129. } else if len(cd.Jobs) > 0 {
  130. name = cd.Jobs[0]
  131. kind = "job"
  132. hasController = true
  133. match := isCron.FindStringSubmatch(name)
  134. if match != nil {
  135. name = match[1]
  136. }
  137. }
  138. return name, kind, hasController
  139. }
  140. func (cm *CostModel) ComputeCostData(start, end time.Time) (map[string]*CostData, error) {
  141. // Cluster ID is specific to the source cluster
  142. clusterID := coreenv.GetClusterID()
  143. cp := cm.Provider
  144. ds := cm.DataSource
  145. mq := ds.Metrics()
  146. // Get Kubernetes data
  147. // Pull pod information from k8s API
  148. podlist := cm.Cache.GetAllPods()
  149. podDeploymentsMapping, err := getPodDeployments(cm.Cache, podlist, clusterID)
  150. if err != nil {
  151. return nil, err
  152. }
  153. podServicesMapping, err := getPodServices(cm.Cache, podlist, clusterID)
  154. if err != nil {
  155. return nil, err
  156. }
  157. namespaceLabelsMapping, err := getNamespaceLabels(cm.Cache, clusterID)
  158. if err != nil {
  159. return nil, err
  160. }
  161. namespaceAnnotationsMapping, err := getNamespaceAnnotations(cm.Cache, clusterID)
  162. if err != nil {
  163. return nil, err
  164. }
  165. // Get metrics data
  166. resRAMUsage, resCPUUsage, resNetZoneRequests, resNetRegionRequests, resNetInternetRequests, resNetNatGatewayRequests, resNetNatGatewayIngressRequests, err := queryMetrics(mq, start, end)
  167. if err != nil {
  168. log.Warnf("ComputeCostData: continuing despite metrics errors: %s", err)
  169. }
  170. defer measureTime(time.Now(), profileThreshold, "ComputeCostData: Processing Query Data")
  171. nodes, err := cm.GetNodeCost()
  172. if err != nil {
  173. log.Warnf("GetNodeCost: no node cost model available: %s", err)
  174. return nil, err
  175. }
  176. // Unmounted PVs represent the PVs that are not mounted or tied to a volume on a container
  177. unmountedPVs := make(map[string][]*PersistentVolumeClaimData)
  178. pvClaimMapping, err := GetPVInfoLocal(cm.Cache, clusterID)
  179. if err != nil {
  180. log.Warnf("GetPVInfo: unable to get PV data: %s", err.Error())
  181. }
  182. if pvClaimMapping != nil {
  183. err = cm.addPVData(pvClaimMapping)
  184. if err != nil {
  185. return nil, err
  186. }
  187. // copy claim mappings into zombies, then remove as they're discovered
  188. for k, v := range pvClaimMapping {
  189. unmountedPVs[k] = []*PersistentVolumeClaimData{v}
  190. }
  191. }
  192. networkUsageMap, err := GetNetworkUsageData(resNetZoneRequests, resNetRegionRequests, resNetInternetRequests, resNetNatGatewayRequests, resNetNatGatewayIngressRequests, clusterID)
  193. if err != nil {
  194. log.Warnf("Unable to get Network Cost Data: %s", err.Error())
  195. networkUsageMap = make(map[string]*NetworkUsageData)
  196. }
  197. containerNameCost := make(map[string]*CostData)
  198. containers := make(map[string]bool)
  199. RAMUsedMap, err := GetContainerMetricVector(resRAMUsage, clusterID)
  200. if err != nil {
  201. return nil, err
  202. }
  203. for key := range RAMUsedMap {
  204. containers[key] = true
  205. }
  206. CPUUsedMap, err := GetContainerMetricVector(resCPUUsage, clusterID) // No need to normalize here, as this comes from a counter
  207. if err != nil {
  208. return nil, err
  209. }
  210. for key := range CPUUsedMap {
  211. containers[key] = true
  212. }
  213. currentContainers := make(map[string]clustercache.Pod)
  214. for _, pod := range podlist {
  215. if pod.Status.Phase != v1.PodRunning {
  216. continue
  217. }
  218. cs, err := NewContainerMetricsFromPod(pod, clusterID)
  219. if err != nil {
  220. return nil, err
  221. }
  222. for _, c := range cs {
  223. containers[c.Key()] = true // captures any containers that existed for a time < a metrics scrape interval. We currently charge 0 for this but should charge something.
  224. currentContainers[c.Key()] = *pod
  225. }
  226. }
  227. missingNodes := make(map[string]*costAnalyzerCloud.Node)
  228. missingContainers := make(map[string]*CostData)
  229. for key := range containers {
  230. if _, ok := containerNameCost[key]; ok {
  231. continue // because ordering is important for the allocation model (all PV's applied to the first), just dedupe if it's already been added.
  232. }
  233. // The _else_ case for this statement is the case in which the container has been
  234. // deleted so we have usage information but not request information. In that case,
  235. // we return partial data for CPU and RAM: only usage and not requests.
  236. if pod, ok := currentContainers[key]; ok {
  237. podName := pod.Name
  238. ns := pod.Namespace
  239. nsLabels := namespaceLabelsMapping[ns+","+clusterID]
  240. podLabels := maps.Clone(pod.Labels)
  241. if podLabels == nil {
  242. podLabels = make(map[string]string)
  243. }
  244. for k, v := range nsLabels {
  245. if _, ok := podLabels[k]; !ok {
  246. podLabels[k] = v
  247. }
  248. }
  249. nsAnnotations := namespaceAnnotationsMapping[ns+","+clusterID]
  250. podAnnotations := pod.Annotations
  251. if podAnnotations == nil {
  252. podAnnotations = make(map[string]string)
  253. }
  254. for k, v := range nsAnnotations {
  255. if _, ok := podAnnotations[k]; !ok {
  256. podAnnotations[k] = v
  257. }
  258. }
  259. nodeName := pod.Spec.NodeName
  260. var nodeData *costAnalyzerCloud.Node
  261. if _, ok := nodes[nodeName]; ok {
  262. nodeData = nodes[nodeName]
  263. }
  264. nsKey := ns + "," + clusterID
  265. var podDeployments []string
  266. if _, ok := podDeploymentsMapping[nsKey]; ok {
  267. if ds, ok := podDeploymentsMapping[nsKey][pod.Name]; ok {
  268. podDeployments = ds
  269. } else {
  270. podDeployments = []string{}
  271. }
  272. }
  273. var podPVs []*PersistentVolumeClaimData
  274. podClaims := pod.Spec.Volumes
  275. for _, vol := range podClaims {
  276. if vol.PersistentVolumeClaim != nil {
  277. name := vol.PersistentVolumeClaim.ClaimName
  278. key := ns + "," + name + "," + clusterID
  279. if pvClaim, ok := pvClaimMapping[key]; ok {
  280. pvClaim.TimesClaimed++
  281. podPVs = append(podPVs, pvClaim)
  282. // Remove entry from potential unmounted pvs
  283. delete(unmountedPVs, key)
  284. }
  285. }
  286. }
  287. var podNetCosts []*util.Vector
  288. if usage, ok := networkUsageMap[ns+","+podName+","+clusterID]; ok {
  289. netCosts, err := GetNetworkCost(usage, cp)
  290. if err != nil {
  291. log.Debugf("Error pulling network costs: %s", err.Error())
  292. } else {
  293. podNetCosts = netCosts
  294. }
  295. }
  296. var podServices []string
  297. if _, ok := podServicesMapping[nsKey]; ok {
  298. if svcs, ok := podServicesMapping[nsKey][pod.Name]; ok {
  299. podServices = svcs
  300. } else {
  301. podServices = []string{}
  302. }
  303. }
  304. for i, container := range pod.Spec.Containers {
  305. containerName := container.Name
  306. // recreate the key and look up data for this container
  307. newKey := NewContainerMetricFromValues(ns, podName, containerName, pod.Spec.NodeName, clusterID).Key()
  308. // k8s.io/apimachinery/pkg/api/resource/amount.go and
  309. // k8s.io/apimachinery/pkg/api/resource/quantity.go for
  310. // details on the "amount" API. See
  311. // https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/#resource-types
  312. // for the units of memory and CPU.
  313. ramRequestBytes := container.Resources.Requests.Memory().Value()
  314. // Because information on container RAM & CPU requests isn't
  315. // coming from metrics, it won't have a timestamp associated
  316. // with it. We need to provide a timestamp.
  317. RAMReqV := []*util.Vector{
  318. {
  319. Value: float64(ramRequestBytes),
  320. Timestamp: float64(time.Now().UTC().Unix()),
  321. },
  322. }
  323. // use millicores so we can convert to cores in a float64 format
  324. cpuRequestMilliCores := container.Resources.Requests.Cpu().MilliValue()
  325. CPUReqV := []*util.Vector{
  326. {
  327. Value: float64(cpuRequestMilliCores) / 1000,
  328. Timestamp: float64(time.Now().UTC().Unix()),
  329. },
  330. }
  331. gpuReqCount := 0.0
  332. if g, ok := container.Resources.Requests["nvidia.com/gpu"]; ok {
  333. gpuReqCount = g.AsApproximateFloat64()
  334. } else if g, ok := container.Resources.Limits["nvidia.com/gpu"]; ok {
  335. gpuReqCount = g.AsApproximateFloat64()
  336. } else if g, ok := container.Resources.Requests["k8s.amazonaws.com/vgpu"]; ok {
  337. gpuReqCount = g.AsApproximateFloat64()
  338. } else if g, ok := container.Resources.Limits["k8s.amazonaws.com/vgpu"]; ok {
  339. gpuReqCount = g.AsApproximateFloat64()
  340. }
  341. GPUReqV := []*util.Vector{
  342. {
  343. Value: float64(gpuReqCount),
  344. Timestamp: float64(time.Now().UTC().Unix()),
  345. },
  346. }
  347. RAMUsedV, ok := RAMUsedMap[newKey]
  348. if !ok {
  349. log.Debug("no RAM usage for " + newKey)
  350. RAMUsedV = []*util.Vector{{}}
  351. }
  352. CPUUsedV, ok := CPUUsedMap[newKey]
  353. if !ok {
  354. log.Debug("no CPU usage for " + newKey)
  355. CPUUsedV = []*util.Vector{{}}
  356. }
  357. var pvReq []*PersistentVolumeClaimData
  358. var netReq []*util.Vector
  359. if i == 0 { // avoid duplicating by just assigning all claims to the first container.
  360. pvReq = podPVs
  361. netReq = podNetCosts
  362. }
  363. costs := &CostData{
  364. Name: containerName,
  365. PodName: podName,
  366. NodeName: nodeName,
  367. Namespace: ns,
  368. Deployments: podDeployments,
  369. Services: podServices,
  370. Daemonsets: getDaemonsetsOfPod(pod),
  371. Jobs: getJobsOfPod(pod),
  372. Statefulsets: getStatefulSetsOfPod(pod),
  373. NodeData: nodeData,
  374. RAMReq: RAMReqV,
  375. RAMUsed: RAMUsedV,
  376. CPUReq: CPUReqV,
  377. CPUUsed: CPUUsedV,
  378. GPUReq: GPUReqV,
  379. PVCData: pvReq,
  380. NetworkData: netReq,
  381. Annotations: podAnnotations,
  382. Labels: podLabels,
  383. NamespaceLabels: nsLabels,
  384. ClusterID: clusterID,
  385. ClusterName: cm.ClusterMap.NameFor(clusterID),
  386. }
  387. var cpuReq, cpuUse *util.Vector
  388. if len(costs.CPUReq) > 0 {
  389. cpuReq = costs.CPUReq[0]
  390. }
  391. if len(costs.CPUUsed) > 0 {
  392. cpuUse = costs.CPUUsed[0]
  393. }
  394. costs.CPUAllocation = getContainerAllocation(cpuReq, cpuUse, "CPU")
  395. var ramReq, ramUse *util.Vector
  396. if len(costs.RAMReq) > 0 {
  397. ramReq = costs.RAMReq[0]
  398. }
  399. if len(costs.RAMUsed) > 0 {
  400. ramUse = costs.RAMUsed[0]
  401. }
  402. costs.RAMAllocation = getContainerAllocation(ramReq, ramUse, "RAM")
  403. containerNameCost[newKey] = costs
  404. }
  405. } else {
  406. // The container has been deleted. Not all information is sent to metrics via ksm, so fill out what we can without k8s api
  407. log.Debug("The container " + key + " has been deleted. Calculating allocation but resulting object will be missing data.")
  408. c, err := NewContainerMetricFromKey(key)
  409. if err != nil {
  410. return nil, err
  411. }
  412. // CPU and RAM requests are obtained from the Kubernetes API.
  413. // If this case has been reached, the Kubernetes API will not
  414. // have information about the pod because it no longer exists.
  415. //
  416. // The case where this matters is minimal, mainly in environments
  417. // with very short-lived pods that over-request resources.
  418. RAMReqV := []*util.Vector{{}}
  419. CPUReqV := []*util.Vector{{}}
  420. GPUReqV := []*util.Vector{{}}
  421. RAMUsedV, ok := RAMUsedMap[key]
  422. if !ok {
  423. log.Debug("no RAM usage for " + key)
  424. RAMUsedV = []*util.Vector{{}}
  425. }
  426. CPUUsedV, ok := CPUUsedMap[key]
  427. if !ok {
  428. log.Debug("no CPU usage for " + key)
  429. CPUUsedV = []*util.Vector{{}}
  430. }
  431. node, ok := nodes[c.NodeName]
  432. if !ok {
  433. log.Debugf("Node \"%s\" has been deleted from Kubernetes. Query historical data to get it.", c.NodeName)
  434. if n, ok := missingNodes[c.NodeName]; ok {
  435. node = n
  436. } else {
  437. node = &costAnalyzerCloud.Node{}
  438. missingNodes[c.NodeName] = node
  439. }
  440. }
  441. namespacelabels := namespaceLabelsMapping[c.Namespace+","+c.ClusterID]
  442. namespaceAnnotations := namespaceAnnotationsMapping[c.Namespace+","+c.ClusterID]
  443. costs := &CostData{
  444. Name: c.ContainerName,
  445. PodName: c.PodName,
  446. NodeName: c.NodeName,
  447. NodeData: node,
  448. Namespace: c.Namespace,
  449. RAMReq: RAMReqV,
  450. RAMUsed: RAMUsedV,
  451. CPUReq: CPUReqV,
  452. CPUUsed: CPUUsedV,
  453. GPUReq: GPUReqV,
  454. Annotations: namespaceAnnotations,
  455. NamespaceLabels: namespacelabels,
  456. ClusterID: c.ClusterID,
  457. ClusterName: cm.ClusterMap.NameFor(c.ClusterID),
  458. }
  459. var cpuReq, cpuUse *util.Vector
  460. if len(costs.CPUReq) > 0 {
  461. cpuReq = costs.CPUReq[0]
  462. }
  463. if len(costs.CPUUsed) > 0 {
  464. cpuUse = costs.CPUUsed[0]
  465. }
  466. costs.CPUAllocation = getContainerAllocation(cpuReq, cpuUse, "CPU")
  467. var ramReq, ramUse *util.Vector
  468. if len(costs.RAMReq) > 0 {
  469. ramReq = costs.RAMReq[0]
  470. }
  471. if len(costs.RAMUsed) > 0 {
  472. ramUse = costs.RAMUsed[0]
  473. }
  474. costs.RAMAllocation = getContainerAllocation(ramReq, ramUse, "RAM")
  475. containerNameCost[key] = costs
  476. missingContainers[key] = costs
  477. }
  478. }
  479. // Use unmounted pvs to create a mapping of "Unmounted-<Namespace>" containers
  480. // to pass along the cost data
  481. unmounted := findUnmountedPVCostData(cm.ClusterMap, unmountedPVs, namespaceLabelsMapping, namespaceAnnotationsMapping)
  482. for k, costs := range unmounted {
  483. log.Debugf("Unmounted PVs in Namespace/ClusterID: %s/%s", costs.Namespace, costs.ClusterID)
  484. containerNameCost[k] = costs
  485. }
  486. err = findDeletedNodeInfo(cm.DataSource, missingNodes, start, end)
  487. if err != nil {
  488. log.Errorf("Error fetching historical node data: %s", err.Error())
  489. }
  490. err = findDeletedPodInfo(cm.DataSource, missingContainers, start, end)
  491. if err != nil {
  492. log.Errorf("Error fetching historical pod data: %s", err.Error())
  493. }
  494. return containerNameCost, err
  495. }
  496. func queryMetrics(mq source.MetricsQuerier, start, end time.Time) ([]*source.ContainerMetricResult, []*source.ContainerMetricResult, []*source.NetZoneGiBResult, []*source.NetRegionGiBResult, []*source.NetInternetGiBResult, []*source.NetNatGatewayGiBResult, []*source.NetNatGatewayIngressGiBResult, error) {
  497. grp := source.NewQueryGroup()
  498. resChRAMUsage := source.WithGroup(grp, mq.QueryRAMUsageAvg(start, end))
  499. resChCPUUsage := source.WithGroup(grp, mq.QueryCPUUsageAvg(start, end))
  500. resChNetZoneRequests := source.WithGroup(grp, mq.QueryNetZoneGiB(start, end))
  501. resChNetRegionRequests := source.WithGroup(grp, mq.QueryNetRegionGiB(start, end))
  502. resChNetInternetRequests := source.WithGroup(grp, mq.QueryNetInternetGiB(start, end))
  503. resChNetNatGatewayEgressRequests := source.WithGroup(grp, mq.QueryNetNatGatewayGiB(start, end))
  504. resChNetNatGatewayIngressRequests := source.WithGroup(grp, mq.QueryNetNatGatewayIngressGiB(start, end))
  505. // Process metrics query results. Handle errors using ctx.Errors.
  506. resRAMUsage, _ := resChRAMUsage.Await()
  507. resCPUUsage, _ := resChCPUUsage.Await()
  508. resNetZoneRequests, _ := resChNetZoneRequests.Await()
  509. resNetRegionRequests, _ := resChNetRegionRequests.Await()
  510. resNetInternetRequests, _ := resChNetInternetRequests.Await()
  511. resNetNatGatewayEgressRequests, _ := resChNetNatGatewayEgressRequests.Await()
  512. resNetNatGatewayIngressRequests, _ := resChNetNatGatewayIngressRequests.Await()
  513. // NOTE: The way we currently handle errors and warnings only early returns if there is an error. Warnings
  514. // NOTE: will not propagate unless coupled with errors.
  515. if grp.HasErrors() {
  516. // To keep the context of where the errors are occurring, we log the errors here and pass them the error
  517. // back to the caller. The caller should handle the specific case where error is an ErrorCollection
  518. for _, queryErr := range grp.Errors() {
  519. if queryErr.Error != nil {
  520. log.Errorf("ComputeCostData: Request Error: %s", queryErr.Error)
  521. }
  522. if queryErr.ParseError != nil {
  523. log.Errorf("ComputeCostData: Parsing Error: %s", queryErr.ParseError)
  524. }
  525. }
  526. // ErrorCollection is an collection of errors wrapped in a single error implementation
  527. // We opt to not return an error for the sake of running as a pure exporter.
  528. return resRAMUsage, resCPUUsage, resNetZoneRequests, resNetRegionRequests, resNetInternetRequests, resNetNatGatewayEgressRequests, resNetNatGatewayIngressRequests, grp.Error()
  529. }
  530. return resRAMUsage, resCPUUsage, resNetZoneRequests, resNetRegionRequests, resNetInternetRequests, resNetNatGatewayEgressRequests, resNetNatGatewayIngressRequests, nil
  531. }
  532. func findUnmountedPVCostData(clusterMap clusters.ClusterMap, unmountedPVs map[string][]*PersistentVolumeClaimData, namespaceLabelsMapping map[string]map[string]string, namespaceAnnotationsMapping map[string]map[string]string) map[string]*CostData {
  533. costs := make(map[string]*CostData)
  534. if len(unmountedPVs) == 0 {
  535. return costs
  536. }
  537. for k, pv := range unmountedPVs {
  538. keyParts := strings.Split(k, ",")
  539. if len(keyParts) != 3 {
  540. log.Warnf("Unmounted PV used key with incorrect parts: %s", k)
  541. continue
  542. }
  543. ns, _, clusterID := keyParts[0], keyParts[1], keyParts[2]
  544. namespacelabels := namespaceLabelsMapping[ns+","+clusterID]
  545. namespaceAnnotations := namespaceAnnotationsMapping[ns+","+clusterID]
  546. metric := NewContainerMetricFromValues(ns, unmountedPVsContainer, unmountedPVsContainer, "", clusterID)
  547. key := metric.Key()
  548. if costData, ok := costs[key]; !ok {
  549. costs[key] = &CostData{
  550. Name: unmountedPVsContainer,
  551. PodName: unmountedPVsContainer,
  552. NodeName: "",
  553. Annotations: namespaceAnnotations,
  554. Namespace: ns,
  555. NamespaceLabels: namespacelabels,
  556. Labels: namespacelabels,
  557. ClusterID: clusterID,
  558. ClusterName: clusterMap.NameFor(clusterID),
  559. PVCData: pv,
  560. }
  561. } else {
  562. costData.PVCData = append(costData.PVCData, pv...)
  563. }
  564. }
  565. return costs
  566. }
  567. func findDeletedPodInfo(dataSource source.OpenCostDataSource, missingContainers map[string]*CostData, start, end time.Time) error {
  568. if len(missingContainers) > 0 {
  569. mq := dataSource.Metrics()
  570. podLabelsResCh := mq.QueryPodLabels(start, end)
  571. podLabelsResult, err := podLabelsResCh.Await()
  572. if err != nil {
  573. log.Errorf("failed to parse historical pod labels: %s", err.Error())
  574. }
  575. podLabels := make(map[string]map[string]string)
  576. if podLabelsResult != nil {
  577. podLabels, err = parsePodLabels(podLabelsResult)
  578. if err != nil {
  579. log.Errorf("failed to parse historical pod labels: %s", err.Error())
  580. }
  581. }
  582. for key, costData := range missingContainers {
  583. cm, _ := NewContainerMetricFromKey(key)
  584. labels, ok := podLabels[cm.PodName]
  585. if !ok {
  586. labels = make(map[string]string)
  587. }
  588. for k, v := range costData.NamespaceLabels {
  589. labels[k] = v
  590. }
  591. costData.Labels = labels
  592. }
  593. }
  594. return nil
  595. }
  596. func findDeletedNodeInfo(dataSource source.OpenCostDataSource, missingNodes map[string]*costAnalyzerCloud.Node, start, end time.Time) error {
  597. if len(missingNodes) > 0 {
  598. defer measureTime(time.Now(), profileThreshold, "Finding Deleted Node Info")
  599. grp := source.NewQueryGroup()
  600. mq := dataSource.Metrics()
  601. cpuCostResCh := source.WithGroup(grp, mq.QueryNodeCPUPricePerHr(start, end))
  602. ramCostResCh := source.WithGroup(grp, mq.QueryNodeRAMPricePerGiBHr(start, end))
  603. gpuCostResCh := source.WithGroup(grp, mq.QueryNodeGPUPricePerHr(start, end))
  604. cpuCostRes, _ := cpuCostResCh.Await()
  605. ramCostRes, _ := ramCostResCh.Await()
  606. gpuCostRes, _ := gpuCostResCh.Await()
  607. if grp.HasErrors() {
  608. return grp.Error()
  609. }
  610. cpuCosts, err := getCost(cpuCostRes, cpuCostNode, cpuCostData)
  611. if err != nil {
  612. return err
  613. }
  614. ramCosts, err := getCost(ramCostRes, ramCostNode, ramCostData)
  615. if err != nil {
  616. return err
  617. }
  618. gpuCosts, err := getCost(gpuCostRes, gpuCostNode, gpuCostData)
  619. if err != nil {
  620. return err
  621. }
  622. if len(cpuCosts) == 0 {
  623. log.Infof("Opencost metrics not currently available. Ingest this server's /metrics endpoint to get that data.")
  624. }
  625. for node, costv := range cpuCosts {
  626. if _, ok := missingNodes[node]; ok {
  627. missingNodes[node].VCPUCost = fmt.Sprintf("%f", costv[0].Value)
  628. } else {
  629. log.DedupedWarningf(5, "Node `%s` in metrics but not k8s api", node)
  630. }
  631. }
  632. for node, costv := range ramCosts {
  633. if _, ok := missingNodes[node]; ok {
  634. missingNodes[node].RAMCost = fmt.Sprintf("%f", costv[0].Value)
  635. }
  636. }
  637. for node, costv := range gpuCosts {
  638. if _, ok := missingNodes[node]; ok {
  639. missingNodes[node].GPUCost = fmt.Sprintf("%f", costv[0].Value)
  640. }
  641. }
  642. }
  643. return nil
  644. }
  645. // getContainerAllocation takes the max between request and usage. This function
  646. // returns a slice containing a single element describing the container's
  647. // allocation.
  648. //
  649. // Additionally, the timestamp of the allocation will be the highest value
  650. // timestamp between the two vectors. This mitigates situations where
  651. // Timestamp=0. This should have no effect on the metrics emitted by the
  652. // CostModelMetricsEmitter
  653. func getContainerAllocation(req *util.Vector, used *util.Vector, allocationType string) []*util.Vector {
  654. var result []*util.Vector
  655. if req != nil && used != nil {
  656. x1 := req.Value
  657. if math.IsNaN(x1) {
  658. log.Debugf("NaN value found during %s allocation calculation for requests.", allocationType)
  659. x1 = 0.0
  660. }
  661. y1 := used.Value
  662. if math.IsNaN(y1) {
  663. log.Debugf("NaN value found during %s allocation calculation for used.", allocationType)
  664. y1 = 0.0
  665. }
  666. result = []*util.Vector{
  667. {
  668. Value: math.Max(x1, y1),
  669. Timestamp: math.Max(req.Timestamp, used.Timestamp),
  670. },
  671. }
  672. if result[0].Value == 0 && result[0].Timestamp == 0 {
  673. log.Debugf("No request or usage data found during %s allocation calculation. Setting allocation to 0.", allocationType)
  674. }
  675. } else if req != nil {
  676. result = []*util.Vector{
  677. {
  678. Value: req.Value,
  679. Timestamp: req.Timestamp,
  680. },
  681. }
  682. } else if used != nil {
  683. result = []*util.Vector{
  684. {
  685. Value: used.Value,
  686. Timestamp: used.Timestamp,
  687. },
  688. }
  689. } else {
  690. log.Debugf("No request or usage data found during %s allocation calculation. Setting allocation to 0.", allocationType)
  691. result = []*util.Vector{
  692. {
  693. Value: 0,
  694. Timestamp: float64(time.Now().UTC().Unix()),
  695. },
  696. }
  697. }
  698. return result
  699. }
  700. func (cm *CostModel) addPVData(pvClaimMapping map[string]*PersistentVolumeClaimData) error {
  701. cache := cm.Cache
  702. cloud := cm.Provider
  703. cfg, err := cloud.GetConfig()
  704. if err != nil {
  705. return err
  706. }
  707. // Pull a region from the first node
  708. var defaultRegion string
  709. nodeList := cache.GetAllNodes()
  710. if len(nodeList) > 0 {
  711. defaultRegion, _ = util.GetRegion(nodeList[0].Labels)
  712. }
  713. storageClasses := cache.GetAllStorageClasses()
  714. storageClassMap := make(map[string]map[string]string)
  715. for _, storageClass := range storageClasses {
  716. params := storageClass.Parameters
  717. storageClassMap[storageClass.Name] = params
  718. if storageClass.Annotations["storageclass.kubernetes.io/is-default-class"] == "true" || storageClass.Annotations["storageclass.beta.kubernetes.io/is-default-class"] == "true" {
  719. storageClassMap["default"] = params
  720. storageClassMap[""] = params
  721. }
  722. }
  723. pvs := cache.GetAllPersistentVolumes()
  724. pvMap := make(map[string]*costAnalyzerCloud.PV)
  725. for _, pv := range pvs {
  726. parameters, ok := storageClassMap[pv.Spec.StorageClassName]
  727. if !ok {
  728. log.Debugf("Unable to find parameters for storage class \"%s\". Does pv \"%s\" have a storageClassName?", pv.Spec.StorageClassName, pv.Name)
  729. }
  730. var region string
  731. if r, ok := util.GetRegion(pv.Labels); ok {
  732. region = r
  733. } else {
  734. region = defaultRegion
  735. }
  736. cacPv := &costAnalyzerCloud.PV{
  737. Class: pv.Spec.StorageClassName,
  738. Region: region,
  739. Parameters: parameters,
  740. }
  741. err := cm.GetPVCost(cacPv, pv, region)
  742. if err != nil {
  743. return err
  744. }
  745. pvMap[pv.Name] = cacPv
  746. }
  747. for _, pvc := range pvClaimMapping {
  748. if vol, ok := pvMap[pvc.VolumeName]; ok {
  749. pvc.Volume = vol
  750. } else {
  751. log.Debugf("PV not found, using default")
  752. pvc.Volume = &costAnalyzerCloud.PV{
  753. Cost: cfg.Storage,
  754. }
  755. }
  756. }
  757. return nil
  758. }
  759. func (cm *CostModel) GetPVCost(pv *costAnalyzerCloud.PV, kpv *clustercache.PersistentVolume, defaultRegion string) error {
  760. cp := cm.Provider
  761. cfg, err := cp.GetConfig()
  762. if err != nil {
  763. return err
  764. }
  765. key := cp.GetPVKey(kpv, pv.Parameters, defaultRegion)
  766. pv.ProviderID = key.ID()
  767. pvWithCost, err := cp.PVPricing(key)
  768. if err != nil {
  769. pv.Cost = cfg.Storage
  770. return err
  771. }
  772. if pvWithCost == nil || pvWithCost.Cost == "" {
  773. pv.Cost = cfg.Storage
  774. return nil // set default cost
  775. }
  776. pv.Cost = pvWithCost.Cost
  777. return nil
  778. }
  779. func (cm *CostModel) GetPricingSourceCounts() (*costAnalyzerCloud.PricingMatchMetadata, error) {
  780. if cm.pricingMetadata != nil {
  781. return cm.pricingMetadata, nil
  782. } else {
  783. return nil, fmt.Errorf("Node costs not yet calculated")
  784. }
  785. }
  786. func (cm *CostModel) GetNodeCost() (map[string]*costAnalyzerCloud.Node, error) {
  787. cp := cm.Provider
  788. cfg, err := cp.GetConfig()
  789. if err != nil {
  790. return nil, err
  791. }
  792. nodeList := cm.Cache.GetAllNodes()
  793. nodes := make(map[string]*costAnalyzerCloud.Node)
  794. pmd := &costAnalyzerCloud.PricingMatchMetadata{
  795. TotalNodes: 0,
  796. PricingTypeCounts: make(map[costAnalyzerCloud.PricingType]int),
  797. }
  798. for _, n := range nodeList {
  799. name := n.Name
  800. nodeLabels := n.Labels
  801. if nodeLabels == nil {
  802. log.Warnf("GetNodeCost: Found node '%s' with no labels", name)
  803. nodeLabels = make(map[string]string)
  804. }
  805. nodeLabels["providerID"] = n.SpecProviderID
  806. pmd.TotalNodes++
  807. cnode, _, err := cp.NodePricing(cp.GetKey(nodeLabels, n))
  808. if err != nil {
  809. log.DedupedInfof(10, "Could not get node pricing for node %s: %s. Falling back to default pricing.", name, err.Error())
  810. if cnode != nil {
  811. nodes[name] = cnode
  812. continue
  813. } else {
  814. cnode = &costAnalyzerCloud.Node{
  815. VCPUCost: cfg.CPU,
  816. RAMCost: cfg.RAM,
  817. }
  818. }
  819. }
  820. pmd.PricingTypeCounts[cnode.PricingType]++
  821. // newCnode builds upon cnode but populates/overrides certain fields.
  822. // cnode was populated leveraging cloud provider public pricing APIs.
  823. newCnode := *cnode
  824. if newCnode.InstanceType == "" {
  825. it, _ := util.GetInstanceType(n.Labels)
  826. newCnode.InstanceType = it
  827. }
  828. if newCnode.Region == "" {
  829. region, _ := util.GetRegion(n.Labels)
  830. newCnode.Region = region
  831. }
  832. if newCnode.ArchType == "" {
  833. arch, _ := util.GetArchType(n.Labels)
  834. newCnode.ArchType = arch
  835. }
  836. newCnode.ProviderID = n.SpecProviderID
  837. var cpu float64
  838. if newCnode.VCPU == "" {
  839. cpu = float64(n.Status.Capacity.Cpu().Value())
  840. newCnode.VCPU = n.Status.Capacity.Cpu().String()
  841. } else {
  842. cpu, err = strconv.ParseFloat(newCnode.VCPU, 64)
  843. if err != nil {
  844. log.Warnf("parsing VCPU value: \"%s\" as float64", newCnode.VCPU)
  845. }
  846. }
  847. if math.IsNaN(cpu) {
  848. log.Warnf("cpu parsed as NaN. Setting to 0.")
  849. cpu = 0
  850. }
  851. if newCnode.RAM == "" {
  852. newCnode.RAM = n.Status.Capacity.Memory().String()
  853. }
  854. if newCnode.RAMBytes == "" {
  855. newCnode.RAMBytes = fmt.Sprintf("%v", n.Status.Capacity.Memory().Value())
  856. }
  857. ram, _ := strconv.ParseFloat(newCnode.RAMBytes, 64)
  858. if math.IsNaN(ram) {
  859. log.Warnf("ram parsed as NaN. Setting to 0.")
  860. ram = 0
  861. }
  862. gpuc, err := strconv.ParseFloat(newCnode.GPU, 64)
  863. if err != nil {
  864. gpuc = 0.0
  865. }
  866. // The k8s API will often report more accurate results for GPU count
  867. // than cloud provider public pricing APIs. If found, override the
  868. // original value.
  869. gpuOverride, vgpuOverride, err := getGPUCount(cm.Cache, n)
  870. if err != nil {
  871. log.Warnf("Unable to get GPUCount for node %s: %s", n.Name, err.Error())
  872. }
  873. if gpuOverride > 0 {
  874. newCnode.GPU = fmt.Sprintf("%f", gpuOverride)
  875. gpuc = gpuOverride
  876. }
  877. if vgpuOverride > 0 {
  878. newCnode.VGPU = fmt.Sprintf("%f", vgpuOverride)
  879. }
  880. // Special case for SUSE rancher, since it won't behave with normal
  881. // calculations, courtesy of the instance type not being "real" (a
  882. // recognizable AWS instance type.)
  883. if newCnode.InstanceType == "rke2" {
  884. log.Infof(
  885. "Found a SUSE Rancher node %s, defaulting and skipping math",
  886. cp.GetKey(nodeLabels, n).Features(),
  887. )
  888. defaultCPUCorePrice, err := strconv.ParseFloat(cfg.CPU, 64)
  889. if err != nil {
  890. log.Errorf("Could not parse default cpu price")
  891. defaultCPUCorePrice = 0
  892. }
  893. if math.IsNaN(defaultCPUCorePrice) {
  894. log.Warnf("defaultCPU parsed as NaN. Setting to 0.")
  895. defaultCPUCorePrice = 0
  896. }
  897. // Some customers may want GPU pricing to be determined by the labels affixed to their nodes. GpuPricing
  898. // passes the node's labels to the provider, which then cross-references them with the labels that the
  899. // provider knows to have label-specific costs associated with them, and returns that cost. See CSVProvider
  900. // for an example implementation.
  901. var gpuPrice float64
  902. gpuPricing, err := cp.GpuPricing(nodeLabels)
  903. if err != nil {
  904. log.Errorf("Could not determine custom GPU pricing: %s", err)
  905. gpuPrice = 0
  906. } else if len(gpuPricing) > 0 {
  907. gpuPrice, err = strconv.ParseFloat(gpuPricing, 64)
  908. if err != nil {
  909. log.Errorf("Could not parse custom GPU pricing: %s", err)
  910. gpuPrice = 0
  911. } else if math.IsNaN(gpuPrice) {
  912. log.Warnf("Custom GPU pricing parsed as NaN. Setting to 0.")
  913. gpuPrice = 0
  914. } else {
  915. log.Infof("Using custom GPU pricing for node \"%s\": %f", name, gpuPrice)
  916. }
  917. } else {
  918. gpuPrice, err = strconv.ParseFloat(cfg.GPU, 64)
  919. if err != nil {
  920. log.Errorf("Could not parse default gpu price")
  921. gpuPrice = 0
  922. }
  923. if math.IsNaN(gpuPrice) {
  924. log.Warnf("defaultGPU parsed as NaN. Setting to 0.")
  925. gpuPrice = 0
  926. }
  927. }
  928. defaultRAMPrice, err := strconv.ParseFloat(cfg.RAM, 64)
  929. if err != nil {
  930. log.Errorf("Could not parse default ram price")
  931. defaultRAMPrice = 0
  932. }
  933. if math.IsNaN(defaultRAMPrice) {
  934. log.Warnf("defaultRAM parsed as NaN. Setting to 0.")
  935. defaultRAMPrice = 0
  936. }
  937. defaultGPUPrice, err := strconv.ParseFloat(cfg.GPU, 64)
  938. if err != nil {
  939. log.Errorf("Could not parse default gpu price")
  940. defaultGPUPrice = 0
  941. }
  942. if math.IsNaN(defaultGPUPrice) {
  943. log.Warnf("defaultGPU parsed as NaN. Setting to 0.")
  944. defaultGPUPrice = 0
  945. }
  946. // Just say no to doing the ratios!
  947. cpuCost := defaultCPUCorePrice * cpu
  948. gpuCost := gpuPrice * gpuc
  949. ramCost := defaultRAMPrice * ram
  950. nodeCost := cpuCost + gpuCost + ramCost
  951. newCnode.Cost = fmt.Sprintf("%f", nodeCost)
  952. newCnode.VCPUCost = fmt.Sprintf("%f", defaultCPUCorePrice)
  953. newCnode.GPUCost = fmt.Sprintf("%f", gpuPrice)
  954. newCnode.RAMCost = fmt.Sprintf("%f", defaultRAMPrice)
  955. newCnode.RAMBytes = fmt.Sprintf("%f", ram)
  956. } else if newCnode.GPU != "" && newCnode.GPUCost == "" {
  957. // was the big thing to investigate. All the funky ratio math
  958. // we were doing was messing with their default pricing. for SUSE Rancher.
  959. // We reach this when a GPU is detected on a node, but no cost for
  960. // the GPU is defined in the OnDemand pricing. Calculate ratios of
  961. // CPU to RAM and GPU to RAM costs, then distribute the total node
  962. // cost among the CPU, RAM, and GPU.
  963. log.Tracef("GPU without cost found for %s, calculating...", cp.GetKey(nodeLabels, n).Features())
  964. // Some customers may want GPU pricing to be determined by the labels affixed to their nodes. GpuPricing
  965. // passes the node's labels to the provider, which then cross-references them with the labels that the
  966. // provider knows to have label-specific costs associated with them, and returns that cost. See CSVProvider
  967. // for an example implementation.
  968. gpuPricing, err := cp.GpuPricing(nodeLabels)
  969. if err != nil {
  970. log.Errorf("Could not determine custom GPU pricing: %s", err)
  971. } else if len(gpuPricing) > 0 {
  972. newCnode.GPUCost = gpuPricing
  973. log.Infof("Using custom GPU pricing for node \"%s\": %s", name, gpuPricing)
  974. }
  975. if newCnode.GPUCost == "" {
  976. defaultCPU, err := strconv.ParseFloat(cfg.CPU, 64)
  977. if err != nil {
  978. log.Errorf("Could not parse default cpu price")
  979. defaultCPU = 0
  980. }
  981. if math.IsNaN(defaultCPU) {
  982. log.Warnf("defaultCPU parsed as NaN. Setting to 0.")
  983. defaultCPU = 0
  984. }
  985. defaultRAM, err := strconv.ParseFloat(cfg.RAM, 64)
  986. if err != nil {
  987. log.Errorf("Could not parse default ram price")
  988. defaultRAM = 0
  989. }
  990. if math.IsNaN(defaultRAM) {
  991. log.Warnf("defaultRAM parsed as NaN. Setting to 0.")
  992. defaultRAM = 0
  993. }
  994. defaultGPU, err := strconv.ParseFloat(cfg.GPU, 64)
  995. if err != nil {
  996. log.Errorf("Could not parse default gpu price")
  997. defaultGPU = 0
  998. }
  999. if math.IsNaN(defaultGPU) {
  1000. log.Warnf("defaultGPU parsed as NaN. Setting to 0.")
  1001. defaultGPU = 0
  1002. }
  1003. cpuToRAMRatio := defaultCPU / defaultRAM
  1004. if math.IsNaN(cpuToRAMRatio) {
  1005. log.Warnf("cpuToRAMRatio[defaultCPU: %f / defaultRAM: %f] is NaN. Setting to 10.", defaultCPU, defaultRAM)
  1006. cpuToRAMRatio = 10
  1007. }
  1008. gpuToRAMRatio := defaultGPU / defaultRAM
  1009. if math.IsNaN(gpuToRAMRatio) {
  1010. log.Warnf("gpuToRAMRatio is NaN. Setting to 100.")
  1011. gpuToRAMRatio = 100
  1012. }
  1013. ramGB := ram / 1024 / 1024 / 1024
  1014. if math.IsNaN(ramGB) {
  1015. log.Warnf("ramGB is NaN. Setting to 0.")
  1016. ramGB = 0
  1017. }
  1018. ramMultiple := gpuc*gpuToRAMRatio + cpu*cpuToRAMRatio + ramGB
  1019. if math.IsNaN(ramMultiple) {
  1020. log.Warnf("ramMultiple is NaN. Setting to 0.")
  1021. ramMultiple = 0
  1022. }
  1023. var nodePrice float64
  1024. if newCnode.Cost != "" {
  1025. nodePrice, err = strconv.ParseFloat(newCnode.Cost, 64)
  1026. if err != nil {
  1027. log.Errorf("Could not parse total node price")
  1028. return nil, err
  1029. }
  1030. } else if newCnode.VCPUCost != "" {
  1031. nodePrice, err = strconv.ParseFloat(newCnode.VCPUCost, 64) // all the price was allocated to the CPU
  1032. if err != nil {
  1033. log.Errorf("Could not parse node vcpu price")
  1034. return nil, err
  1035. }
  1036. } else { // add case to use default pricing model when API data fails.
  1037. log.Debugf("No node price or CPUprice found, falling back to default")
  1038. nodePrice = defaultCPU*cpu + defaultRAM*ram + gpuc*defaultGPU
  1039. }
  1040. if math.IsNaN(nodePrice) {
  1041. log.Warnf("nodePrice parsed as NaN. Setting to 0.")
  1042. nodePrice = 0
  1043. }
  1044. ramPrice := (nodePrice / ramMultiple)
  1045. if math.IsNaN(ramPrice) {
  1046. log.Warnf("ramPrice[nodePrice: %f / ramMultiple: %f] parsed as NaN. Setting to 0.", nodePrice, ramMultiple)
  1047. ramPrice = 0
  1048. }
  1049. cpuPrice := ramPrice * cpuToRAMRatio
  1050. gpuPrice := ramPrice * gpuToRAMRatio
  1051. newCnode.VCPUCost = fmt.Sprintf("%f", cpuPrice)
  1052. newCnode.RAMCost = fmt.Sprintf("%f", ramPrice)
  1053. newCnode.RAMBytes = fmt.Sprintf("%f", ram)
  1054. newCnode.GPUCost = fmt.Sprintf("%f", gpuPrice)
  1055. }
  1056. } else if newCnode.RAMCost == "" {
  1057. // We reach this when no RAM cost is defined in the OnDemand
  1058. // pricing. It calculates a cpuToRAMRatio and ramMultiple to
  1059. // distrubte the total node cost among CPU and RAM costs.
  1060. log.Tracef("No RAM cost found for %s, calculating...", cp.GetKey(nodeLabels, n).Features())
  1061. defaultCPU, err := strconv.ParseFloat(cfg.CPU, 64)
  1062. if err != nil {
  1063. log.Warnf("Could not parse default cpu price")
  1064. defaultCPU = 0
  1065. }
  1066. if math.IsNaN(defaultCPU) {
  1067. log.Warnf("defaultCPU parsed as NaN. Setting to 0.")
  1068. defaultCPU = 0
  1069. }
  1070. defaultRAM, err := strconv.ParseFloat(cfg.RAM, 64)
  1071. if err != nil {
  1072. log.Warnf("Could not parse default ram price")
  1073. defaultRAM = 0
  1074. }
  1075. if math.IsNaN(defaultRAM) {
  1076. log.Warnf("defaultRAM parsed as NaN. Setting to 0.")
  1077. defaultRAM = 0
  1078. }
  1079. cpuToRAMRatio := defaultCPU / defaultRAM
  1080. if math.IsNaN(cpuToRAMRatio) {
  1081. log.Warnf("cpuToRAMRatio[defaultCPU: %f / defaultRAM: %f] is NaN. Setting to 10.", defaultCPU, defaultRAM)
  1082. cpuToRAMRatio = 10
  1083. }
  1084. ramGB := ram / 1024 / 1024 / 1024
  1085. if math.IsNaN(ramGB) {
  1086. log.Warnf("ramGB is NaN. Setting to 0.")
  1087. ramGB = 0
  1088. }
  1089. ramMultiple := cpu*cpuToRAMRatio + ramGB
  1090. if math.IsNaN(ramMultiple) {
  1091. log.Warnf("ramMultiple is NaN. Setting to 0.")
  1092. ramMultiple = 0
  1093. }
  1094. var nodePrice float64
  1095. if newCnode.Cost != "" {
  1096. nodePrice, err = strconv.ParseFloat(newCnode.Cost, 64)
  1097. if err != nil {
  1098. log.Warnf("Could not parse total node price")
  1099. return nil, err
  1100. }
  1101. if newCnode.GPUCost != "" {
  1102. gpuPrice, err := strconv.ParseFloat(newCnode.GPUCost, 64)
  1103. if err != nil {
  1104. log.Warnf("Could not parse node gpu price")
  1105. return nil, err
  1106. }
  1107. nodePrice = nodePrice - gpuPrice // remove the gpuPrice from the total, we're just costing out RAM and CPU.
  1108. }
  1109. } else if newCnode.VCPUCost != "" {
  1110. nodePrice, err = strconv.ParseFloat(newCnode.VCPUCost, 64) // all the price was allocated to the CPU
  1111. if err != nil {
  1112. log.Warnf("Could not parse node vcpu price")
  1113. return nil, err
  1114. }
  1115. } else { // add case to use default pricing model when API data fails.
  1116. log.Debugf("No node price or CPUprice found, falling back to default")
  1117. nodePrice = defaultCPU*cpu + defaultRAM*ramGB
  1118. }
  1119. if math.IsNaN(nodePrice) {
  1120. log.Warnf("nodePrice parsed as NaN. Setting to 0.")
  1121. nodePrice = 0
  1122. }
  1123. ramPrice := (nodePrice / ramMultiple)
  1124. if math.IsNaN(ramPrice) {
  1125. log.Warnf("ramPrice[nodePrice: %f / ramMultiple: %f] parsed as NaN. Setting to 0.", nodePrice, ramMultiple)
  1126. ramPrice = 0
  1127. }
  1128. cpuPrice := ramPrice * cpuToRAMRatio
  1129. if defaultRAM != 0 {
  1130. newCnode.VCPUCost = fmt.Sprintf("%f", cpuPrice)
  1131. newCnode.RAMCost = fmt.Sprintf("%f", ramPrice)
  1132. } else { // just assign the full price to CPU
  1133. if cpu != 0 {
  1134. newCnode.VCPUCost = fmt.Sprintf("%f", nodePrice/cpu)
  1135. } else {
  1136. newCnode.VCPUCost = fmt.Sprintf("%f", nodePrice)
  1137. }
  1138. }
  1139. newCnode.RAMBytes = fmt.Sprintf("%f", ram)
  1140. log.Tracef("Computed \"%s\" RAM Cost := %v", name, newCnode.RAMCost)
  1141. }
  1142. nodes[name] = &newCnode
  1143. }
  1144. cm.pricingMetadata = pmd
  1145. cp.ApplyReservedInstancePricing(nodes)
  1146. return nodes, nil
  1147. }
  1148. // TODO: drop some logs
  1149. func (cm *CostModel) GetLBCost() (map[serviceKey]*costAnalyzerCloud.LoadBalancer, error) {
  1150. // for fetching prices from cloud provider
  1151. // cfg, err := cp.GetConfig()
  1152. // if err != nil {
  1153. // return nil, err
  1154. // }
  1155. cp := cm.Provider
  1156. servicesList := cm.Cache.GetAllServices()
  1157. loadBalancerMap := make(map[serviceKey]*costAnalyzerCloud.LoadBalancer)
  1158. for _, service := range servicesList {
  1159. namespace := service.Namespace
  1160. name := service.Name
  1161. key := serviceKey{
  1162. Cluster: coreenv.GetClusterID(),
  1163. Namespace: namespace,
  1164. Service: name,
  1165. }
  1166. if service.Type == "LoadBalancer" {
  1167. loadBalancer, err := cp.LoadBalancerPricing()
  1168. if err != nil {
  1169. return nil, err
  1170. }
  1171. newLoadBalancer := *loadBalancer
  1172. for _, loadBalancerIngress := range service.Status.LoadBalancer.Ingress {
  1173. address := loadBalancerIngress.IP
  1174. // Some cloud providers use hostname rather than IP
  1175. if address == "" {
  1176. address = loadBalancerIngress.Hostname
  1177. }
  1178. newLoadBalancer.IngressIPAddresses = append(newLoadBalancer.IngressIPAddresses, address)
  1179. }
  1180. loadBalancerMap[key] = &newLoadBalancer
  1181. }
  1182. }
  1183. return loadBalancerMap, nil
  1184. }
  1185. func getPodServices(cache clustercache.ClusterCache, podList []*clustercache.Pod, clusterID string) (map[string]map[string][]string, error) {
  1186. servicesList := cache.GetAllServices()
  1187. podServicesMapping := make(map[string]map[string][]string)
  1188. for _, service := range servicesList {
  1189. namespace := service.Namespace
  1190. name := service.Name
  1191. key := namespace + "," + clusterID
  1192. if _, ok := podServicesMapping[key]; !ok {
  1193. podServicesMapping[key] = make(map[string][]string)
  1194. }
  1195. s := labels.Nothing()
  1196. if len(service.SpecSelector) > 0 {
  1197. s = labels.Set(service.SpecSelector).AsSelectorPreValidated()
  1198. }
  1199. for _, pod := range podList {
  1200. labelSet := labels.Set(pod.Labels)
  1201. if s.Matches(labelSet) && pod.Namespace == namespace {
  1202. services, ok := podServicesMapping[key][pod.Name]
  1203. if ok {
  1204. podServicesMapping[key][pod.Name] = append(services, name)
  1205. } else {
  1206. podServicesMapping[key][pod.Name] = []string{name}
  1207. }
  1208. }
  1209. }
  1210. }
  1211. return podServicesMapping, nil
  1212. }
  1213. func getPodStatefulsets(cache clustercache.ClusterCache, podList []*clustercache.Pod, clusterID string) (map[string]map[string][]string, error) {
  1214. ssList := cache.GetAllStatefulSets()
  1215. podSSMapping := make(map[string]map[string][]string) // namespace: podName: [deploymentNames]
  1216. for _, ss := range ssList {
  1217. namespace := ss.Namespace
  1218. name := ss.Name
  1219. key := namespace + "," + clusterID
  1220. if _, ok := podSSMapping[key]; !ok {
  1221. podSSMapping[key] = make(map[string][]string)
  1222. }
  1223. s, err := metav1.LabelSelectorAsSelector(ss.SpecSelector)
  1224. if err != nil {
  1225. log.Errorf("Error doing deployment label conversion: %s", err.Error())
  1226. }
  1227. for _, pod := range podList {
  1228. labelSet := labels.Set(pod.Labels)
  1229. if s.Matches(labelSet) && pod.Namespace == namespace {
  1230. sss, ok := podSSMapping[key][pod.Name]
  1231. if ok {
  1232. podSSMapping[key][pod.Name] = append(sss, name)
  1233. } else {
  1234. podSSMapping[key][pod.Name] = []string{name}
  1235. }
  1236. }
  1237. }
  1238. }
  1239. return podSSMapping, nil
  1240. }
  1241. func getPodDeployments(cache clustercache.ClusterCache, podList []*clustercache.Pod, clusterID string) (map[string]map[string][]string, error) {
  1242. deploymentsList := cache.GetAllDeployments()
  1243. podDeploymentsMapping := make(map[string]map[string][]string) // namespace: podName: [deploymentNames]
  1244. for _, deployment := range deploymentsList {
  1245. namespace := deployment.Namespace
  1246. name := deployment.Name
  1247. key := namespace + "," + clusterID
  1248. if _, ok := podDeploymentsMapping[key]; !ok {
  1249. podDeploymentsMapping[key] = make(map[string][]string)
  1250. }
  1251. s, err := metav1.LabelSelectorAsSelector(deployment.SpecSelector)
  1252. if err != nil {
  1253. log.Errorf("Error doing deployment label conversion: %s", err)
  1254. }
  1255. for _, pod := range podList {
  1256. labelSet := labels.Set(pod.Labels)
  1257. if s.Matches(labelSet) && pod.Namespace == namespace {
  1258. deployments, ok := podDeploymentsMapping[key][pod.Name]
  1259. if ok {
  1260. podDeploymentsMapping[key][pod.Name] = append(deployments, name)
  1261. } else {
  1262. podDeploymentsMapping[key][pod.Name] = []string{name}
  1263. }
  1264. }
  1265. }
  1266. }
  1267. return podDeploymentsMapping, nil
  1268. }
  1269. func getNamespaceLabels(cache clustercache.ClusterCache, clusterID string) (map[string]map[string]string, error) {
  1270. nsToLabels := make(map[string]map[string]string)
  1271. nss := cache.GetAllNamespaces()
  1272. for _, ns := range nss {
  1273. labels := make(map[string]string)
  1274. for k, v := range ns.Labels {
  1275. labels[promutil.SanitizeLabelName(k)] = v
  1276. }
  1277. nsToLabels[ns.Name+","+clusterID] = labels
  1278. }
  1279. return nsToLabels, nil
  1280. }
  1281. func getNamespaceAnnotations(cache clustercache.ClusterCache, clusterID string) (map[string]map[string]string, error) {
  1282. nsToAnnotations := make(map[string]map[string]string)
  1283. nss := cache.GetAllNamespaces()
  1284. for _, ns := range nss {
  1285. annotations := make(map[string]string)
  1286. for k, v := range ns.Annotations {
  1287. annotations[promutil.SanitizeLabelName(k)] = v
  1288. }
  1289. nsToAnnotations[ns.Name+","+clusterID] = annotations
  1290. }
  1291. return nsToAnnotations, nil
  1292. }
  1293. func getDaemonsetsOfPod(pod clustercache.Pod) []string {
  1294. for _, ownerReference := range pod.OwnerReferences {
  1295. if ownerReference.Kind == "DaemonSet" {
  1296. return []string{ownerReference.Name}
  1297. }
  1298. }
  1299. return []string{}
  1300. }
  1301. func getJobsOfPod(pod clustercache.Pod) []string {
  1302. for _, ownerReference := range pod.OwnerReferences {
  1303. if ownerReference.Kind == "Job" {
  1304. return []string{ownerReference.Name}
  1305. }
  1306. }
  1307. return []string{}
  1308. }
  1309. func getStatefulSetsOfPod(pod clustercache.Pod) []string {
  1310. for _, ownerReference := range pod.OwnerReferences {
  1311. if ownerReference.Kind == "StatefulSet" {
  1312. return []string{ownerReference.Name}
  1313. }
  1314. }
  1315. return []string{}
  1316. }
  1317. // getGPUCount reads the node's Status and Labels (via the k8s API) to identify
  1318. // the number of GPUs and vGPUs are equipped on the node. If unable to identify
  1319. // a GPU count, it will return -1.
  1320. func getGPUCount(cache clustercache.ClusterCache, n *clustercache.Node) (float64, float64, error) {
  1321. g, hasGpu := n.Status.Capacity["nvidia.com/gpu"]
  1322. _, hasReplicas := n.Labels["nvidia.com/gpu.replicas"]
  1323. // Case 1: Standard NVIDIA GPU
  1324. if hasGpu && g.Value() != 0 && !hasReplicas {
  1325. return float64(g.Value()), float64(g.Value()), nil
  1326. }
  1327. // Case 2: NVIDIA GPU with GPU Feature Discovery (GFD) Pod enabled.
  1328. // Ref: https://docs.nvidia.com/datacenter/cloud-native/gpu-operator/latest/gpu-sharing.html#verifying-the-gpu-time-slicing-configuration
  1329. // Ref: https://github.com/NVIDIA/k8s-device-plugin/blob/d899752a424818428f744a946d32b132ea2c0cf1/internal/lm/resource_test.go#L44-L45
  1330. // Ref: https://github.com/NVIDIA/k8s-device-plugin/blob/d899752a424818428f744a946d32b132ea2c0cf1/internal/lm/resource_test.go#L103-L118
  1331. if hasReplicas {
  1332. resultGPU := 0.0
  1333. resultVGPU := 0.0
  1334. if c, ok := n.Labels["nvidia.com/gpu.count"]; ok {
  1335. var err error
  1336. resultGPU, err = strconv.ParseFloat(c, 64)
  1337. if err != nil {
  1338. return -1, -1, fmt.Errorf("could not parse label \"nvidia.com/gpu.count\": %v", err)
  1339. }
  1340. }
  1341. if s, ok := n.Status.Capacity["nvidia.com/gpu.shared"]; ok { // GFD configured `renameByDefault=true`
  1342. resultVGPU = float64(s.Value())
  1343. } else if g, ok := n.Status.Capacity["nvidia.com/gpu"]; ok { // GFD configured `renameByDefault=false`
  1344. resultVGPU = float64(g.Value())
  1345. } else {
  1346. resultVGPU = resultGPU
  1347. }
  1348. return resultGPU, resultVGPU, nil
  1349. }
  1350. // Case 3: AWS vGPU
  1351. if vgpu, ok := n.Status.Capacity["k8s.amazonaws.com/vgpu"]; ok {
  1352. vgpuCount, err := getAllocatableVGPUs(cache)
  1353. if err != nil {
  1354. return -1, -1, err
  1355. }
  1356. vgpuCoeff := 10.0
  1357. if vgpuCount > 0.0 {
  1358. vgpuCoeff = vgpuCount
  1359. }
  1360. if vgpu.Value() != 0 {
  1361. resultGPU := float64(vgpu.Value()) / vgpuCoeff
  1362. resultVGPU := float64(vgpu.Value())
  1363. return resultGPU, resultVGPU, nil
  1364. }
  1365. }
  1366. // No GPU found
  1367. return -1, -1, nil
  1368. }
  1369. func getAllocatableVGPUs(cache clustercache.ClusterCache) (float64, error) {
  1370. daemonsets := cache.GetAllDaemonSets()
  1371. vgpuCount := 0.0
  1372. for _, ds := range daemonsets {
  1373. dsContainerList := &ds.SpecContainers
  1374. for _, ctnr := range *dsContainerList {
  1375. if ctnr.Args != nil {
  1376. for _, arg := range ctnr.Args {
  1377. if strings.Contains(arg, "--vgpu=") {
  1378. vgpus, err := strconv.ParseFloat(arg[strings.IndexByte(arg, '=')+1:], 64)
  1379. if err != nil {
  1380. log.Errorf("failed to parse vgpu allocation string %s: %v", arg, err)
  1381. continue
  1382. }
  1383. vgpuCount = vgpus
  1384. return vgpuCount, nil
  1385. }
  1386. }
  1387. }
  1388. }
  1389. }
  1390. return vgpuCount, nil
  1391. }
  1392. type PersistentVolumeClaimData struct {
  1393. Class string `json:"class"`
  1394. Claim string `json:"claim"`
  1395. Namespace string `json:"namespace"`
  1396. ClusterID string `json:"clusterId"`
  1397. TimesClaimed int `json:"timesClaimed"`
  1398. VolumeName string `json:"volumeName"`
  1399. Volume *costAnalyzerCloud.PV `json:"persistentVolume"`
  1400. Values []*util.Vector `json:"values"`
  1401. }
  1402. func measureTime(start time.Time, threshold time.Duration, name string) {
  1403. elapsed := time.Since(start)
  1404. if elapsed > threshold {
  1405. log.Infof("[Profiler] %s: %s", elapsed, name)
  1406. }
  1407. }
  1408. func (cm *CostModel) QueryAllocation(window opencost.Window, step time.Duration, aggregate []string, includeIdle, idleByNode, includeProportionalAssetResourceCosts, includeAggregatedMetadata, sharedLoadBalancer bool, accumulateBy opencost.AccumulateOption, shareIdle bool, filterString string) (*opencost.AllocationSetRange, error) {
  1409. // Validate window is legal
  1410. if window.IsOpen() || window.IsNegative() {
  1411. return nil, fmt.Errorf("illegal window: %s", window)
  1412. }
  1413. var totalsStore opencost.TotalsStore
  1414. // Idle is required for proportional asset costs
  1415. if includeProportionalAssetResourceCosts {
  1416. if !includeIdle {
  1417. return nil, errors.New("bad request - includeIdle must be set true if includeProportionalAssetResourceCosts is true")
  1418. }
  1419. totalsStore = opencost.NewMemoryTotalsStore()
  1420. }
  1421. // Begin with empty response
  1422. asr := opencost.NewAllocationSetRange()
  1423. // Query for AllocationSets in increments of the given step duration,
  1424. // appending each to the response.
  1425. stepStart := *window.Start()
  1426. stepEnd := stepStart.Add(step)
  1427. var isAKS bool
  1428. for window.End().After(stepStart) {
  1429. allocSet, err := cm.ComputeAllocation(stepStart, stepEnd)
  1430. if err != nil {
  1431. return nil, fmt.Errorf("error computing allocations for %s: %w", opencost.NewClosedWindow(stepStart, stepEnd), err)
  1432. }
  1433. if includeIdle {
  1434. assetSet, err := cm.ComputeAssets(stepStart, stepEnd)
  1435. if err != nil {
  1436. return nil, fmt.Errorf("error computing assets for %s: %w", opencost.NewClosedWindow(stepStart, stepEnd), err)
  1437. }
  1438. if includeProportionalAssetResourceCosts {
  1439. // AKS is a special case - there can be a maximum of 2
  1440. // load balancers (1 public and 1 private) in an AKS cluster
  1441. // therefore, when calculating PARCs for load balancers,
  1442. // we must know if this is an AKS cluster
  1443. for _, node := range assetSet.Nodes {
  1444. if _, found := node.Labels["label_kubernetes_azure_com_cluster"]; found {
  1445. isAKS = true
  1446. break
  1447. }
  1448. }
  1449. _, err := opencost.UpdateAssetTotalsStore(totalsStore, assetSet)
  1450. if err != nil {
  1451. log.Errorf("Allocation: error updating asset resource totals for %s: %s", assetSet.Window, err)
  1452. }
  1453. }
  1454. idleSet, err := computeIdleAllocations(allocSet, assetSet, idleByNode)
  1455. if err != nil {
  1456. return nil, fmt.Errorf("error computing idle allocations for %s: %w", opencost.NewClosedWindow(stepStart, stepEnd), err)
  1457. }
  1458. for _, idleAlloc := range idleSet.Allocations {
  1459. allocSet.Insert(idleAlloc)
  1460. }
  1461. }
  1462. asr.Append(allocSet)
  1463. stepStart = stepEnd
  1464. stepEnd = stepStart.Add(step)
  1465. }
  1466. // Apply allocation filter BEFORE aggregation if provided
  1467. if filterString != "" {
  1468. parser := allocation.NewAllocationFilterParser()
  1469. filterNode, err := parser.Parse(filterString)
  1470. if err != nil {
  1471. return nil, fmt.Errorf("invalid filter: %w", err)
  1472. }
  1473. compiler := opencost.NewAllocationMatchCompiler(nil)
  1474. matcher, err := compiler.Compile(filterNode)
  1475. if err != nil {
  1476. return nil, fmt.Errorf("failed to compile filter: %w", err)
  1477. }
  1478. filteredASR := opencost.NewAllocationSetRange()
  1479. for _, as := range asr.Slice() {
  1480. filteredAS := opencost.NewAllocationSet(as.Start(), as.End())
  1481. for _, alloc := range as.Allocations {
  1482. if matcher.Matches(alloc) {
  1483. filteredAS.Set(alloc)
  1484. }
  1485. }
  1486. if filteredAS.Length() > 0 {
  1487. filteredASR.Append(filteredAS)
  1488. }
  1489. }
  1490. asr = filteredASR
  1491. }
  1492. // Set aggregation options and aggregate
  1493. var shareIdleOpt string
  1494. if shareIdle {
  1495. shareIdleOpt = opencost.ShareWeighted
  1496. } else {
  1497. shareIdleOpt = opencost.ShareNone
  1498. }
  1499. opts := &opencost.AllocationAggregationOptions{
  1500. IncludeProportionalAssetResourceCosts: includeProportionalAssetResourceCosts,
  1501. IdleByNode: idleByNode,
  1502. IncludeAggregatedMetadata: includeAggregatedMetadata,
  1503. ShareIdle: shareIdleOpt,
  1504. }
  1505. // Aggregate
  1506. err := asr.AggregateBy(aggregate, opts)
  1507. if err != nil {
  1508. return nil, fmt.Errorf("error aggregating for %s: %w", window, err)
  1509. }
  1510. // Accumulate, if requested
  1511. if accumulateBy != opencost.AccumulateOptionNone {
  1512. asr, err = asr.Accumulate(accumulateBy)
  1513. if err != nil {
  1514. log.Errorf("error accumulating by %v: %s", accumulateBy, err)
  1515. return nil, fmt.Errorf("error accumulating by %v: %s", accumulateBy, err)
  1516. }
  1517. // when accumulating and returning PARCs, we need the totals for the
  1518. // accumulated windows to accurately compute a fraction
  1519. if includeProportionalAssetResourceCosts {
  1520. assetSet, err := cm.ComputeAssets(*asr.Window().Start(), *asr.Window().End())
  1521. if err != nil {
  1522. return nil, fmt.Errorf("error computing assets for %s: %w", opencost.NewClosedWindow(*asr.Window().Start(), *asr.Window().End()), err)
  1523. }
  1524. _, err = opencost.UpdateAssetTotalsStore(totalsStore, assetSet)
  1525. if err != nil {
  1526. log.Errorf("Allocation: error updating asset resource totals for %s: %s", opencost.NewClosedWindow(*asr.Window().Start(), *asr.Window().End()), err)
  1527. }
  1528. }
  1529. }
  1530. if includeProportionalAssetResourceCosts {
  1531. for _, as := range asr.Allocations {
  1532. totalStoreByNode, ok := totalsStore.GetAssetTotalsByNode(as.Start(), as.End())
  1533. if !ok {
  1534. log.Errorf("unable to locate allocation totals for node for window %v - %v", as.Start(), as.End())
  1535. return nil, fmt.Errorf("unable to locate allocation totals for node for window %v - %v", as.Start(), as.End())
  1536. }
  1537. totalStoreByCluster, ok := totalsStore.GetAssetTotalsByCluster(as.Start(), as.End())
  1538. if !ok {
  1539. log.Errorf("unable to locate allocation totals for cluster for window %v - %v", as.Start(), as.End())
  1540. return nil, fmt.Errorf("unable to locate allocation totals for cluster for window %v - %v", as.Start(), as.End())
  1541. }
  1542. var totalPublicLbCost, totalPrivateLbCost float64
  1543. if isAKS && sharedLoadBalancer {
  1544. // loop through all assetTotals, adding all load balancer costs by public and private
  1545. for _, tot := range totalStoreByNode {
  1546. if tot.PrivateLoadBalancer {
  1547. totalPrivateLbCost += tot.LoadBalancerCost
  1548. } else {
  1549. totalPublicLbCost += tot.LoadBalancerCost
  1550. }
  1551. }
  1552. }
  1553. // loop through each allocation set, using total cost from totals store
  1554. for _, alloc := range as.Allocations {
  1555. for rawKey, parc := range alloc.ProportionalAssetResourceCosts {
  1556. key := strings.TrimSuffix(strings.ReplaceAll(rawKey, ",", "/"), "/")
  1557. // for each parc , check the totals store for each
  1558. // on a totals hit, set the corresponding total and calculate percentage
  1559. var totals *opencost.AssetTotals
  1560. if totalsLoc, found := totalStoreByCluster[key]; found {
  1561. totals = totalsLoc
  1562. }
  1563. if totalsLoc, found := totalStoreByNode[key]; found {
  1564. totals = totalsLoc
  1565. }
  1566. if totals == nil {
  1567. log.Errorf("unable to locate asset totals for allocation %s, corresponding PARC is being skipped", key)
  1568. continue
  1569. }
  1570. parc.CPUTotalCost = totals.CPUCost
  1571. parc.GPUTotalCost = totals.GPUCost
  1572. parc.RAMTotalCost = totals.RAMCost
  1573. parc.PVTotalCost = totals.PersistentVolumeCost
  1574. if isAKS && sharedLoadBalancer && len(alloc.LoadBalancers) > 0 {
  1575. // Azure is a special case - use computed totals above
  1576. // use the lbAllocations in the object to determine if
  1577. // this PARC is a public or private load balancer
  1578. // then set the total accordingly
  1579. // AKS only has 1 public and 1 private load balancer
  1580. lbAlloc, found := alloc.LoadBalancers[key]
  1581. if found {
  1582. if lbAlloc.Private {
  1583. parc.LoadBalancerTotalCost = totalPrivateLbCost
  1584. } else {
  1585. parc.LoadBalancerTotalCost = totalPublicLbCost
  1586. }
  1587. }
  1588. } else {
  1589. parc.LoadBalancerTotalCost = totals.LoadBalancerCost
  1590. }
  1591. opencost.ComputePercentages(&parc)
  1592. alloc.ProportionalAssetResourceCosts[rawKey] = parc
  1593. }
  1594. }
  1595. }
  1596. }
  1597. return asr, nil
  1598. }
  1599. // debugAssetAllocationMismatch analyzes and logs discrepancies between asset and allocation data
  1600. // This helps diagnose pricing issues and negative idle costs
  1601. func debugAssetAllocationMismatch(allocSet *opencost.AllocationSet, assetSet *opencost.AssetSet) {
  1602. log.Debugf("=== Asset-Allocation Debug Analysis for window %s ===", allocSet.Window)
  1603. // Build maps for efficient lookup
  1604. assetsByProviderID := make(map[string]*opencost.Node)
  1605. assetsByNode := make(map[string]*opencost.Node)
  1606. for _, asset := range assetSet.Nodes {
  1607. if asset.Properties != nil && asset.Properties.ProviderID != "" {
  1608. assetsByProviderID[asset.Properties.ProviderID] = asset
  1609. }
  1610. if asset.Properties != nil && asset.Properties.Name != "" {
  1611. assetsByNode[asset.Properties.Name] = asset
  1612. }
  1613. }
  1614. // 1) Find allocations without matching assets (by ProviderID)
  1615. allocsWithoutAssets := make([]*opencost.Allocation, 0)
  1616. for _, alloc := range allocSet.Allocations {
  1617. if alloc.Properties == nil {
  1618. continue
  1619. }
  1620. providerID := alloc.Properties.ProviderID
  1621. if providerID == "" {
  1622. continue
  1623. }
  1624. if _, found := assetsByProviderID[providerID]; !found {
  1625. allocsWithoutAssets = append(allocsWithoutAssets, alloc)
  1626. }
  1627. }
  1628. if len(allocsWithoutAssets) > 0 {
  1629. log.Debugf("Found %d allocations without matching assets:", len(allocsWithoutAssets))
  1630. for _, alloc := range allocsWithoutAssets {
  1631. log.Debugf(" - Allocation: %s, Node: %s, ProviderID: %s, TotalCost: %.4f",
  1632. alloc.Name,
  1633. alloc.Properties.Node,
  1634. alloc.Properties.ProviderID,
  1635. alloc.TotalCost())
  1636. }
  1637. }
  1638. // 2) Sum allocations per node and compare to node asset costs
  1639. allocTotalsByNode := make(map[string]*struct {
  1640. CPUCost float64
  1641. GPUCost float64
  1642. RAMCost float64
  1643. TotalCost float64
  1644. CPUCoreHours float64
  1645. GPUHours float64
  1646. RAMByteHours float64
  1647. Count int
  1648. })
  1649. for _, alloc := range allocSet.Allocations {
  1650. if alloc.Properties == nil || alloc.Properties.Node == "" {
  1651. continue
  1652. }
  1653. node := alloc.Properties.Node
  1654. if _, exists := allocTotalsByNode[node]; !exists {
  1655. allocTotalsByNode[node] = &struct {
  1656. CPUCost float64
  1657. GPUCost float64
  1658. RAMCost float64
  1659. TotalCost float64
  1660. CPUCoreHours float64
  1661. GPUHours float64
  1662. RAMByteHours float64
  1663. Count int
  1664. }{}
  1665. }
  1666. allocTotalsByNode[node].CPUCost += alloc.CPUCost
  1667. allocTotalsByNode[node].GPUCost += alloc.GPUCost
  1668. allocTotalsByNode[node].RAMCost += alloc.RAMCost
  1669. allocTotalsByNode[node].TotalCost += alloc.TotalCost()
  1670. allocTotalsByNode[node].CPUCoreHours += alloc.CPUCoreHours
  1671. allocTotalsByNode[node].GPUHours += alloc.GPUHours
  1672. allocTotalsByNode[node].RAMByteHours += alloc.RAMByteHours
  1673. allocTotalsByNode[node].Count++
  1674. }
  1675. log.Debugf("Per-Node Asset vs Allocation Comparison:")
  1676. for node, allocTotals := range allocTotalsByNode {
  1677. asset, hasAsset := assetsByNode[node]
  1678. if !hasAsset {
  1679. log.Debugf(" Node %s: Has allocations but NO ASSET (allocations: %d, total cost: %.4f)",
  1680. node, allocTotals.Count, allocTotals.TotalCost)
  1681. continue
  1682. }
  1683. assetCPU := asset.CPUCost
  1684. assetGPU := asset.GPUCost
  1685. assetRAM := asset.RAMCost
  1686. assetTotal := asset.TotalCost()
  1687. cpuDiff := assetCPU - allocTotals.CPUCost
  1688. gpuDiff := assetGPU - allocTotals.GPUCost
  1689. ramDiff := assetRAM - allocTotals.RAMCost
  1690. totalDiff := assetTotal - allocTotals.TotalCost
  1691. status := "OK"
  1692. if cpuDiff < 0 || gpuDiff < 0 || ramDiff < 0 {
  1693. status = "NEGATIVE_IDLE"
  1694. }
  1695. log.Debugf(" Node %s [%s]:", node, status)
  1696. log.Debugf(" Asset: CPU=%.4f, GPU=%.4f, RAM=%.4f, Total=%.4f",
  1697. assetCPU, assetGPU, assetRAM, assetTotal)
  1698. log.Debugf(" Allocation: CPU=%.4f, GPU=%.4f, RAM=%.4f, Total=%.4f (%d allocs)",
  1699. allocTotals.CPUCost, allocTotals.GPUCost, allocTotals.RAMCost, allocTotals.TotalCost, allocTotals.Count)
  1700. log.Debugf(" Difference: CPU=%.4f, GPU=%.4f, RAM=%.4f, Total=%.4f",
  1701. cpuDiff, gpuDiff, ramDiff, totalDiff)
  1702. if asset.Adjustment != 0 {
  1703. log.Debugf(" Adjustment: %.4f", asset.Adjustment)
  1704. }
  1705. // Compare resource amounts vs costs: higher resources should have higher costs
  1706. assetCPUHours := asset.CPUCoreHours
  1707. assetGPUHours := asset.GPUHours
  1708. assetRAMBytes := asset.RAMByteHours
  1709. allocCPUHours := allocTotals.CPUCoreHours
  1710. allocGPUHours := allocTotals.GPUHours
  1711. allocRAMBytes := allocTotals.RAMByteHours
  1712. // Warn if resource amounts and costs are inverted (higher resources but lower costs)
  1713. if assetCPUHours > 0 && allocCPUHours > 0 {
  1714. if assetCPUHours > allocCPUHours && assetCPU < allocTotals.CPUCost {
  1715. log.Warnf("Resource-cost inversion for %s CPU: asset has MORE hours (%.2f) but LESS cost (%.4f) than allocations (hours: %.2f, cost: %.4f)",
  1716. node, assetCPUHours, assetCPU, allocCPUHours, allocTotals.CPUCost)
  1717. } else if assetCPUHours < allocCPUHours && assetCPU > allocTotals.CPUCost {
  1718. log.Warnf("Resource-cost inversion for %s CPU: asset has LESS hours (%.2f) but MORE cost (%.4f) than allocations (hours: %.2f, cost: %.4f)",
  1719. node, assetCPUHours, assetCPU, allocCPUHours, allocTotals.CPUCost)
  1720. }
  1721. }
  1722. if assetGPUHours > 0 && allocGPUHours > 0 {
  1723. if assetGPUHours > allocGPUHours && assetGPU < allocTotals.GPUCost {
  1724. log.Warnf("Resource-cost inversion for %s GPU: asset has MORE hours (%.2f) but LESS cost (%.4f) than allocations (hours: %.2f, cost: %.4f)",
  1725. node, assetGPUHours, assetGPU, allocGPUHours, allocTotals.GPUCost)
  1726. } else if assetGPUHours < allocGPUHours && assetGPU > allocTotals.GPUCost {
  1727. log.Warnf("Resource-cost inversion for %s GPU: asset has LESS hours (%.2f) but MORE cost (%.4f) than allocations (hours: %.2f, cost: %.4f)",
  1728. node, assetGPUHours, assetGPU, allocGPUHours, allocTotals.GPUCost)
  1729. }
  1730. }
  1731. if assetRAMBytes > 0 && allocRAMBytes > 0 {
  1732. if assetRAMBytes > allocRAMBytes && assetRAM < allocTotals.RAMCost {
  1733. log.Warnf("Resource-cost inversion for %s RAM: asset has MORE byte-hours (%.2f) but LESS cost (%.4f) than allocations (byte-hours: %.2f, cost: %.4f)",
  1734. node, assetRAMBytes, assetRAM, allocRAMBytes, allocTotals.RAMCost)
  1735. } else if assetRAMBytes < allocRAMBytes && assetRAM > allocTotals.RAMCost {
  1736. log.Warnf("Resource-cost inversion for %s RAM: asset has LESS byte-hours (%.2f) but MORE cost (%.4f) than allocations (byte-hours: %.2f, cost: %.4f)",
  1737. node, assetRAMBytes, assetRAM, allocRAMBytes, allocTotals.RAMCost)
  1738. }
  1739. }
  1740. // Log resource amounts for debugging
  1741. log.Debugf(" Resource Hours:")
  1742. log.Debugf(" Asset: CPU=%.2f hours, GPU=%.2f hours, RAM=%.2f byte-hours",
  1743. assetCPUHours, assetGPUHours, assetRAMBytes)
  1744. log.Debugf(" Allocation: CPU=%.2f hours, GPU=%.2f hours, RAM=%.2f byte-hours",
  1745. allocCPUHours, allocGPUHours, allocRAMBytes)
  1746. }
  1747. // 3) Sum total of all node costs
  1748. totalNodeCPU := 0.0
  1749. totalNodeGPU := 0.0
  1750. totalNodeRAM := 0.0
  1751. totalNodeCost := 0.0
  1752. nodeCount := 0
  1753. for _, asset := range assetSet.Nodes {
  1754. totalNodeCPU += asset.CPUCost
  1755. totalNodeGPU += asset.GPUCost
  1756. totalNodeRAM += asset.RAMCost
  1757. totalNodeCost += asset.TotalCost()
  1758. nodeCount++
  1759. }
  1760. log.Debugf("Total Node Asset Costs:")
  1761. log.Debugf(" Nodes: %d", nodeCount)
  1762. log.Debugf(" CPU: %.4f", totalNodeCPU)
  1763. log.Debugf(" GPU: %.4f", totalNodeGPU)
  1764. log.Debugf(" RAM: %.4f", totalNodeRAM)
  1765. log.Debugf(" Total: %.4f", totalNodeCost)
  1766. // 4) Sum total of all allocation costs
  1767. totalAllocCPU := 0.0
  1768. totalAllocGPU := 0.0
  1769. totalAllocRAM := 0.0
  1770. totalAllocCost := 0.0
  1771. allocCount := 0
  1772. for _, alloc := range allocSet.Allocations {
  1773. totalAllocCPU += alloc.CPUCost
  1774. totalAllocGPU += alloc.GPUCost
  1775. totalAllocRAM += alloc.RAMCost
  1776. totalAllocCost += alloc.TotalCost()
  1777. allocCount++
  1778. }
  1779. log.Debugf("Total Allocation Costs:")
  1780. log.Debugf(" Allocations: %d", allocCount)
  1781. log.Debugf(" CPU: %.4f", totalAllocCPU)
  1782. log.Debugf(" GPU: %.4f", totalAllocGPU)
  1783. log.Debugf(" RAM: %.4f", totalAllocRAM)
  1784. log.Debugf(" Total: %.4f", totalAllocCost)
  1785. // Overall comparison
  1786. log.Debugf("Overall Asset vs Allocation:")
  1787. log.Debugf(" CPU Difference: %.4f (Asset - Allocation)", totalNodeCPU-totalAllocCPU)
  1788. log.Debugf(" GPU Difference: %.4f (Asset - Allocation)", totalNodeGPU-totalAllocGPU)
  1789. log.Debugf(" RAM Difference: %.4f (Asset - Allocation)", totalNodeRAM-totalAllocRAM)
  1790. log.Debugf(" Total Difference: %.4f (Asset - Allocation)", totalNodeCost-totalAllocCost)
  1791. log.Debugf("=== End Asset-Allocation Debug Analysis ===")
  1792. }
  1793. func computeIdleAllocations(allocSet *opencost.AllocationSet, assetSet *opencost.AssetSet, idleByNode bool) (*opencost.AllocationSet, error) {
  1794. if !allocSet.Window.Equal(assetSet.Window) {
  1795. return nil, fmt.Errorf("cannot compute idle allocations for mismatched sets: %s does not equal %s", allocSet.Window, assetSet.Window)
  1796. }
  1797. // Run debug analysis when log level is debug
  1798. debugAssetAllocationMismatch(allocSet, assetSet)
  1799. var allocTotals map[string]*opencost.AllocationTotals
  1800. var assetTotals map[string]*opencost.AssetTotals
  1801. if idleByNode {
  1802. allocTotals = opencost.ComputeAllocationTotals(allocSet, opencost.AllocationNodeProp)
  1803. assetTotals = opencost.ComputeAssetTotals(assetSet, true)
  1804. } else {
  1805. allocTotals = opencost.ComputeAllocationTotals(allocSet, opencost.AllocationClusterProp)
  1806. assetTotals = opencost.ComputeAssetTotals(assetSet, false)
  1807. }
  1808. start, end := *allocSet.Window.Start(), *allocSet.Window.End()
  1809. idleSet := opencost.NewAllocationSet(start, end)
  1810. for key, assetTotal := range assetTotals {
  1811. allocTotal, ok := allocTotals[key]
  1812. if !ok {
  1813. log.Warnf("Allocation: did not find allocations for asset key: %s", key)
  1814. // Use a zero-value set of totals. This indicates either (1) an
  1815. // error computing totals, or (2) that no allocations ran on the
  1816. // given node for the given window.
  1817. allocTotal = &opencost.AllocationTotals{
  1818. Cluster: assetTotal.Cluster,
  1819. Node: assetTotal.Node,
  1820. Start: assetTotal.Start,
  1821. End: assetTotal.End,
  1822. }
  1823. }
  1824. // Insert one idle allocation for each key (whether by node or
  1825. // by cluster), defined as the difference between the total
  1826. // asset cost and the allocated cost per-resource.
  1827. // Idle costs are clamped to zero to prevent negative values that can occur
  1828. // when asset total costs are less than allocated costs. This can happen when:
  1829. // - Pricing data is unavailable (promless mode, API failures, missing price data)
  1830. // - Custom pricing is misconfigured or returns zero values
  1831. // - Cloud billing adjustments reduce asset costs below allocation costs
  1832. // - Allocation calculations exceed asset costs due to timing or rounding
  1833. name := fmt.Sprintf("%s/%s", key, opencost.IdleSuffix)
  1834. cpuIdleCost := assetTotal.TotalCPUCost() - allocTotal.TotalCPUCost()
  1835. gpuIdleCost := assetTotal.TotalGPUCost() - allocTotal.TotalGPUCost()
  1836. ramIdleCost := assetTotal.TotalRAMCost() - allocTotal.TotalRAMCost()
  1837. // Clamp idle costs to zero to prevent negative idle allocations
  1838. if cpuIdleCost < 0 {
  1839. log.Warnf("Negative CPU idle cost detected for %s: asset total (%.4f) < allocation total (%.4f), clamping to 0",
  1840. key, assetTotal.TotalCPUCost(), allocTotal.TotalCPUCost())
  1841. cpuIdleCost = 0
  1842. }
  1843. if gpuIdleCost < 0 {
  1844. log.Warnf("Negative GPU idle cost detected for %s: asset total (%.4f) < allocation total (%.4f), clamping to 0",
  1845. key, assetTotal.TotalGPUCost(), allocTotal.TotalGPUCost())
  1846. gpuIdleCost = 0
  1847. }
  1848. if ramIdleCost < 0 {
  1849. log.Warnf("Negative RAM idle cost detected for %s: asset total (%.4f) < allocation total (%.4f), clamping to 0",
  1850. key, assetTotal.TotalRAMCost(), allocTotal.TotalRAMCost())
  1851. ramIdleCost = 0
  1852. }
  1853. err := idleSet.Insert(&opencost.Allocation{
  1854. Name: name,
  1855. Window: idleSet.Window.Clone(),
  1856. Properties: &opencost.AllocationProperties{
  1857. Cluster: assetTotal.Cluster,
  1858. Node: assetTotal.Node,
  1859. ProviderID: assetTotal.ProviderID,
  1860. },
  1861. Start: assetTotal.Start,
  1862. End: assetTotal.End,
  1863. CPUCost: cpuIdleCost,
  1864. GPUCost: gpuIdleCost,
  1865. RAMCost: ramIdleCost,
  1866. })
  1867. if err != nil {
  1868. return nil, fmt.Errorf("failed to insert idle allocation %s: %w", name, err)
  1869. }
  1870. }
  1871. return idleSet, nil
  1872. }
  1873. func (cm *CostModel) GetDataSource() source.OpenCostDataSource {
  1874. return cm.DataSource
  1875. }