costmodel.go 72 KB

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