costmodel.go 77 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227
  1. package costmodel
  2. import (
  3. "fmt"
  4. "math"
  5. "regexp"
  6. "strconv"
  7. "strings"
  8. "time"
  9. costAnalyzerCloud "github.com/kubecost/cost-model/pkg/cloud"
  10. "github.com/kubecost/cost-model/pkg/clustercache"
  11. "github.com/kubecost/cost-model/pkg/costmodel/clusters"
  12. "github.com/kubecost/cost-model/pkg/env"
  13. "github.com/kubecost/cost-model/pkg/kubecost"
  14. "github.com/kubecost/cost-model/pkg/log"
  15. "github.com/kubecost/cost-model/pkg/prom"
  16. "github.com/kubecost/cost-model/pkg/util"
  17. prometheus "github.com/prometheus/client_golang/api"
  18. prometheusClient "github.com/prometheus/client_golang/api"
  19. v1 "k8s.io/api/core/v1"
  20. metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
  21. "k8s.io/apimachinery/pkg/labels"
  22. "k8s.io/klog"
  23. "golang.org/x/sync/singleflight"
  24. )
  25. const (
  26. statusAPIError = 422
  27. profileThreshold = 1000 * 1000 * 1000 // 1s (in ns)
  28. apiPrefix = "/api/v1"
  29. epAlertManagers = apiPrefix + "/alertmanagers"
  30. epLabelValues = apiPrefix + "/label/:name/values"
  31. epSeries = apiPrefix + "/series"
  32. epTargets = apiPrefix + "/targets"
  33. epSnapshot = apiPrefix + "/admin/tsdb/snapshot"
  34. epDeleteSeries = apiPrefix + "/admin/tsdb/delete_series"
  35. epCleanTombstones = apiPrefix + "/admin/tsdb/clean_tombstones"
  36. epConfig = apiPrefix + "/status/config"
  37. epFlags = apiPrefix + "/status/flags"
  38. )
  39. // isCron matches a CronJob name and captures the non-timestamp name
  40. var isCron = regexp.MustCompile(`^(.+)-\d{10}$`)
  41. type CostModel struct {
  42. Cache clustercache.ClusterCache
  43. ClusterMap clusters.ClusterMap
  44. RequestGroup *singleflight.Group
  45. ScrapeInterval time.Duration
  46. PrometheusClient prometheus.Client
  47. Provider costAnalyzerCloud.Provider
  48. pricingMetadata *costAnalyzerCloud.PricingMatchMetadata
  49. }
  50. func NewCostModel(client prometheus.Client, provider costAnalyzerCloud.Provider, cache clustercache.ClusterCache, clusterMap clusters.ClusterMap, scrapeInterval time.Duration) *CostModel {
  51. // request grouping to prevent over-requesting the same data prior to caching
  52. requestGroup := new(singleflight.Group)
  53. return &CostModel{
  54. Cache: cache,
  55. ClusterMap: clusterMap,
  56. PrometheusClient: client,
  57. Provider: provider,
  58. RequestGroup: requestGroup,
  59. ScrapeInterval: scrapeInterval,
  60. }
  61. }
  62. type CostData struct {
  63. Name string `json:"name,omitempty"`
  64. PodName string `json:"podName,omitempty"`
  65. NodeName string `json:"nodeName,omitempty"`
  66. NodeData *costAnalyzerCloud.Node `json:"node,omitempty"`
  67. Namespace string `json:"namespace,omitempty"`
  68. Deployments []string `json:"deployments,omitempty"`
  69. Services []string `json:"services,omitempty"`
  70. Daemonsets []string `json:"daemonsets,omitempty"`
  71. Statefulsets []string `json:"statefulsets,omitempty"`
  72. Jobs []string `json:"jobs,omitempty"`
  73. RAMReq []*util.Vector `json:"ramreq,omitempty"`
  74. RAMUsed []*util.Vector `json:"ramused,omitempty"`
  75. RAMAllocation []*util.Vector `json:"ramallocated,omitempty"`
  76. CPUReq []*util.Vector `json:"cpureq,omitempty"`
  77. CPUUsed []*util.Vector `json:"cpuused,omitempty"`
  78. CPUAllocation []*util.Vector `json:"cpuallocated,omitempty"`
  79. GPUReq []*util.Vector `json:"gpureq,omitempty"`
  80. PVCData []*PersistentVolumeClaimData `json:"pvcData,omitempty"`
  81. NetworkData []*util.Vector `json:"network,omitempty"`
  82. Annotations map[string]string `json:"annotations,omitempty"`
  83. Labels map[string]string `json:"labels,omitempty"`
  84. NamespaceLabels map[string]string `json:"namespaceLabels,omitempty"`
  85. ClusterID string `json:"clusterId"`
  86. ClusterName string `json:"clusterName"`
  87. }
  88. func (cd *CostData) String() string {
  89. 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",
  90. cd.Name, cd.PodName, cd.NodeName, cd.Namespace, strings.Join(cd.Deployments, ", "), strings.Join(cd.Services, ", "),
  91. len(cd.CPUReq), len(cd.CPUUsed), len(cd.CPUAllocation),
  92. len(cd.RAMReq), len(cd.RAMUsed), len(cd.RAMAllocation))
  93. }
  94. func (cd *CostData) GetController() (name string, kind string, hasController bool) {
  95. hasController = false
  96. if len(cd.Deployments) > 0 {
  97. name = cd.Deployments[0]
  98. kind = "deployment"
  99. hasController = true
  100. } else if len(cd.Statefulsets) > 0 {
  101. name = cd.Statefulsets[0]
  102. kind = "statefulset"
  103. hasController = true
  104. } else if len(cd.Daemonsets) > 0 {
  105. name = cd.Daemonsets[0]
  106. kind = "daemonset"
  107. hasController = true
  108. } else if len(cd.Jobs) > 0 {
  109. name = cd.Jobs[0]
  110. kind = "job"
  111. hasController = true
  112. match := isCron.FindStringSubmatch(name)
  113. if match != nil {
  114. name = match[1]
  115. }
  116. }
  117. return name, kind, hasController
  118. }
  119. const (
  120. queryRAMRequestsStr = `avg(
  121. label_replace(
  122. label_replace(
  123. avg(
  124. count_over_time(kube_pod_container_resource_requests_memory_bytes{container!="",container!="POD", node!=""}[%s] %s)
  125. *
  126. avg_over_time(kube_pod_container_resource_requests_memory_bytes{container!="",container!="POD", node!=""}[%s] %s)
  127. ) by (namespace,container,pod,node,cluster_id) , "container_name","$1","container","(.+)"
  128. ), "pod_name","$1","pod","(.+)"
  129. )
  130. ) by (namespace,container_name,pod_name,node,cluster_id)`
  131. queryRAMUsageStr = `sort_desc(
  132. avg(
  133. label_replace(count_over_time(container_memory_working_set_bytes{container_name!="",container_name!="POD", instance!=""}[%s] %s), "node", "$1", "instance","(.+)")
  134. *
  135. label_replace(avg_over_time(container_memory_working_set_bytes{container_name!="",container_name!="POD", instance!=""}[%s] %s), "node", "$1", "instance","(.+)")
  136. ) by (namespace,container_name,pod_name,node,cluster_id)
  137. )`
  138. queryCPURequestsStr = `avg(
  139. label_replace(
  140. label_replace(
  141. avg(
  142. count_over_time(kube_pod_container_resource_requests_cpu_cores{container!="",container!="POD", node!=""}[%s] %s)
  143. *
  144. avg_over_time(kube_pod_container_resource_requests_cpu_cores{container!="",container!="POD", node!=""}[%s] %s)
  145. ) by (namespace,container,pod,node,cluster_id) , "container_name","$1","container","(.+)"
  146. ), "pod_name","$1","pod","(.+)"
  147. )
  148. ) by (namespace,container_name,pod_name,node,cluster_id)`
  149. queryCPUUsageStr = `avg(
  150. label_replace(
  151. rate(
  152. container_cpu_usage_seconds_total{container_name!="",container_name!="POD",instance!=""}[%s] %s
  153. ) , "node", "$1", "instance", "(.+)"
  154. )
  155. ) by (namespace,container_name,pod_name,node,cluster_id)`
  156. queryGPURequestsStr = `avg(
  157. label_replace(
  158. label_replace(
  159. avg(
  160. count_over_time(kube_pod_container_resource_requests{resource="nvidia_com_gpu", container!="",container!="POD", node!=""}[%s] %s)
  161. *
  162. avg_over_time(kube_pod_container_resource_requests{resource="nvidia_com_gpu", container!="",container!="POD", node!=""}[%s] %s)
  163. * %f
  164. ) by (namespace,container,pod,node,cluster_id) , "container_name","$1","container","(.+)"
  165. ), "pod_name","$1","pod","(.+)"
  166. )
  167. ) by (namespace,container_name,pod_name,node,cluster_id)
  168. * on (pod_name, namespace, cluster_id) group_left(container) label_replace(avg(avg_over_time(kube_pod_status_phase{phase="Running"}[%s] %s)) by (pod,namespace,cluster_id), "pod_name","$1","pod","(.+)")`
  169. queryPVRequestsStr = `avg(avg(kube_persistentvolumeclaim_info{volumename != ""}) by (persistentvolumeclaim, storageclass, namespace, volumename, cluster_id, kubernetes_node)
  170. *
  171. on (persistentvolumeclaim, namespace, cluster_id, kubernetes_node) group_right(storageclass, volumename)
  172. sum(kube_persistentvolumeclaim_resource_requests_storage_bytes{}) by (persistentvolumeclaim, namespace, cluster_id, kubernetes_node, kubernetes_name)) by (persistentvolumeclaim, storageclass, namespace, cluster_id, volumename, kubernetes_node)`
  173. // queryRAMAllocationByteHours yields the total byte-hour RAM allocation over the given
  174. // window, aggregated by container.
  175. // [line 3] sum_over_time(each byte) = [byte*scrape] by metric
  176. // [line 4] (scalar(avg(prometheus_target_interval_length_seconds)) = [seconds/scrape] / 60 / 60 = [hours/scrape] by container
  177. // [lines 2,4] sum(") by unique container key and multiply [byte*scrape] * [hours/scrape] for byte*hours
  178. // [lines 1,5] relabeling
  179. queryRAMAllocationByteHours = `
  180. label_replace(label_replace(
  181. sum(
  182. sum_over_time(container_memory_allocation_bytes{container!="",container!="POD", node!=""}[%s])
  183. ) by (namespace,container,pod,node,cluster_id) * %f / 60 / 60
  184. , "container_name","$1","container","(.+)"), "pod_name","$1","pod","(.+)")`
  185. // queryCPUAllocationVCPUHours yields the total VCPU-hour CPU allocation over the given
  186. // window, aggregated by container.
  187. // [line 3] sum_over_time(each VCPU*mins in window) = [VCPU*scrape] by metric
  188. // [line 4] (scalar(avg(prometheus_target_interval_length_seconds)) = [seconds/scrape] / 60 / 60 = [hours/scrape] by container
  189. // [lines 2,4] sum(") by unique container key and multiply [VCPU*scrape] * [hours/scrape] for VCPU*hours
  190. // [lines 1,5] relabeling
  191. queryCPUAllocationVCPUHours = `
  192. label_replace(label_replace(
  193. sum(
  194. sum_over_time(container_cpu_allocation{container!="",container!="POD", node!=""}[%s])
  195. ) by (namespace,container,pod,node,cluster_id) * %f / 60 / 60
  196. , "container_name","$1","container","(.+)"), "pod_name","$1","pod","(.+)")`
  197. // queryPVCAllocationFmt yields the total byte-hour PVC allocation over the given window.
  198. // sum_over_time(each byte) = [byte*scrape] by metric *(scalar(avg(prometheus_target_interval_length_seconds)) = [seconds/scrape] / 60 / 60 = [hours/scrape] by pod
  199. queryPVCAllocationFmt = `sum(sum_over_time(pod_pvc_allocation[%s])) by (cluster_id, namespace, pod, persistentvolume, persistentvolumeclaim) * %f/60/60`
  200. queryPVHourlyCostFmt = `avg_over_time(pv_hourly_cost[%s])`
  201. queryNSLabels = `avg_over_time(kube_namespace_labels[%s])`
  202. queryPodLabels = `avg_over_time(kube_pod_labels[%s])`
  203. queryNSAnnotations = `avg_over_time(kube_namespace_annotations[%s])`
  204. queryPodAnnotations = `avg_over_time(kube_pod_annotations[%s])`
  205. queryDeploymentLabels = `avg_over_time(deployment_match_labels[%s])`
  206. queryStatefulsetLabels = `avg_over_time(statefulSet_match_labels[%s])`
  207. queryPodDaemonsets = `sum(kube_pod_owner{owner_kind="DaemonSet"}) by (namespace,pod,owner_name,cluster_id)`
  208. queryPodJobs = `sum(kube_pod_owner{owner_kind="Job"}) by (namespace,pod,owner_name,cluster_id)`
  209. queryServiceLabels = `avg_over_time(service_selector_labels[%s])`
  210. queryZoneNetworkUsage = `sum(increase(kubecost_pod_network_egress_bytes_total{internet="false", sameZone="false", sameRegion="true"}[%s] %s)) by (namespace,pod_name,cluster_id) / 1024 / 1024 / 1024`
  211. queryRegionNetworkUsage = `sum(increase(kubecost_pod_network_egress_bytes_total{internet="false", sameZone="false", sameRegion="false"}[%s] %s)) by (namespace,pod_name,cluster_id) / 1024 / 1024 / 1024`
  212. queryInternetNetworkUsage = `sum(increase(kubecost_pod_network_egress_bytes_total{internet="true"}[%s] %s)) by (namespace,pod_name,cluster_id) / 1024 / 1024 / 1024`
  213. normalizationStr = `max(count_over_time(kube_pod_container_resource_requests_memory_bytes{}[%s] %s))`
  214. )
  215. func (cm *CostModel) ComputeCostData(cli prometheusClient.Client, cp costAnalyzerCloud.Provider, window string, offset string, filterNamespace string) (map[string]*CostData, error) {
  216. queryRAMUsage := fmt.Sprintf(queryRAMUsageStr, window, offset, window, offset)
  217. queryCPUUsage := fmt.Sprintf(queryCPUUsageStr, window, offset)
  218. queryGPURequests := fmt.Sprintf(queryGPURequestsStr, window, offset, window, offset, 1.0, window, offset)
  219. queryPVRequests := fmt.Sprintf(queryPVRequestsStr)
  220. queryNetZoneRequests := fmt.Sprintf(queryZoneNetworkUsage, window, "")
  221. queryNetRegionRequests := fmt.Sprintf(queryRegionNetworkUsage, window, "")
  222. queryNetInternetRequests := fmt.Sprintf(queryInternetNetworkUsage, window, "")
  223. queryNormalization := fmt.Sprintf(normalizationStr, window, offset)
  224. // Cluster ID is specific to the source cluster
  225. clusterID := env.GetClusterID()
  226. // Submit all Prometheus queries asynchronously
  227. ctx := prom.NewContext(cli)
  228. resChRAMUsage := ctx.Query(queryRAMUsage)
  229. resChCPUUsage := ctx.Query(queryCPUUsage)
  230. resChGPURequests := ctx.Query(queryGPURequests)
  231. resChPVRequests := ctx.Query(queryPVRequests)
  232. resChNetZoneRequests := ctx.Query(queryNetZoneRequests)
  233. resChNetRegionRequests := ctx.Query(queryNetRegionRequests)
  234. resChNetInternetRequests := ctx.Query(queryNetInternetRequests)
  235. resChNormalization := ctx.Query(queryNormalization)
  236. // Pull pod information from k8s API
  237. podlist := cm.Cache.GetAllPods()
  238. podDeploymentsMapping, err := getPodDeployments(cm.Cache, podlist, clusterID)
  239. if err != nil {
  240. return nil, err
  241. }
  242. podServicesMapping, err := getPodServices(cm.Cache, podlist, clusterID)
  243. if err != nil {
  244. return nil, err
  245. }
  246. namespaceLabelsMapping, err := getNamespaceLabels(cm.Cache, clusterID)
  247. if err != nil {
  248. return nil, err
  249. }
  250. namespaceAnnotationsMapping, err := getNamespaceAnnotations(cm.Cache, clusterID)
  251. if err != nil {
  252. return nil, err
  253. }
  254. // Process Prometheus query results. Handle errors using ctx.Errors.
  255. resRAMUsage, _ := resChRAMUsage.Await()
  256. resCPUUsage, _ := resChCPUUsage.Await()
  257. resGPURequests, _ := resChGPURequests.Await()
  258. resPVRequests, _ := resChPVRequests.Await()
  259. resNetZoneRequests, _ := resChNetZoneRequests.Await()
  260. resNetRegionRequests, _ := resChNetRegionRequests.Await()
  261. resNetInternetRequests, _ := resChNetInternetRequests.Await()
  262. resNormalization, _ := resChNormalization.Await()
  263. // NOTE: The way we currently handle errors and warnings only early returns if there is an error. Warnings
  264. // NOTE: will not propagate unless coupled with errors.
  265. if ctx.HasErrors() {
  266. // To keep the context of where the errors are occurring, we log the errors here and pass them the error
  267. // back to the caller. The caller should handle the specific case where error is an ErrorCollection
  268. for _, promErr := range ctx.Errors() {
  269. if promErr.Error != nil {
  270. log.Errorf("ComputeCostData: Request Error: %s", promErr.Error)
  271. }
  272. if promErr.ParseError != nil {
  273. log.Errorf("ComputeCostData: Parsing Error: %s", promErr.ParseError)
  274. }
  275. }
  276. // ErrorCollection is an collection of errors wrapped in a single error implementation
  277. // We opt to not return an error for the sake of running as a pure exporter.
  278. log.Warningf("ComputeCostData: continuing despite prometheus errors: %s", ctx.ErrorCollection().Error())
  279. }
  280. defer measureTime(time.Now(), profileThreshold, "ComputeCostData: Processing Query Data")
  281. normalizationValue, err := getNormalization(resNormalization)
  282. if err != nil {
  283. // We opt to not return an error for the sake of running as a pure exporter.
  284. log.Warningf("ComputeCostData: continuing despite error parsing normalization values from %s: %s", queryNormalization, err.Error())
  285. }
  286. nodes, err := cm.GetNodeCost(cp)
  287. if err != nil {
  288. log.Warningf("GetNodeCost: no node cost model available: " + err.Error())
  289. return nil, err
  290. }
  291. // Unmounted PVs represent the PVs that are not mounted or tied to a volume on a container
  292. unmountedPVs := make(map[string][]*PersistentVolumeClaimData)
  293. pvClaimMapping, err := GetPVInfo(resPVRequests, clusterID)
  294. if err != nil {
  295. log.Warningf("GetPVInfo: unable to get PV data: %s", err.Error())
  296. }
  297. if pvClaimMapping != nil {
  298. err = addPVData(cm.Cache, pvClaimMapping, cp)
  299. if err != nil {
  300. return nil, err
  301. }
  302. // copy claim mappings into zombies, then remove as they're discovered
  303. for k, v := range pvClaimMapping {
  304. unmountedPVs[k] = []*PersistentVolumeClaimData{v}
  305. }
  306. }
  307. networkUsageMap, err := GetNetworkUsageData(resNetZoneRequests, resNetRegionRequests, resNetInternetRequests, clusterID)
  308. if err != nil {
  309. klog.V(1).Infof("[Warning] Unable to get Network Cost Data: %s", err.Error())
  310. networkUsageMap = make(map[string]*NetworkUsageData)
  311. }
  312. containerNameCost := make(map[string]*CostData)
  313. containers := make(map[string]bool)
  314. RAMUsedMap, err := GetContainerMetricVector(resRAMUsage, true, normalizationValue, clusterID)
  315. if err != nil {
  316. return nil, err
  317. }
  318. for key := range RAMUsedMap {
  319. containers[key] = true
  320. }
  321. GPUReqMap, err := GetContainerMetricVector(resGPURequests, true, normalizationValue, clusterID)
  322. if err != nil {
  323. return nil, err
  324. }
  325. for key := range GPUReqMap {
  326. containers[key] = true
  327. }
  328. CPUUsedMap, err := GetContainerMetricVector(resCPUUsage, false, 0, clusterID) // No need to normalize here, as this comes from a counter
  329. if err != nil {
  330. return nil, err
  331. }
  332. for key := range CPUUsedMap {
  333. containers[key] = true
  334. }
  335. currentContainers := make(map[string]v1.Pod)
  336. for _, pod := range podlist {
  337. if pod.Status.Phase != v1.PodRunning {
  338. continue
  339. }
  340. cs, err := NewContainerMetricsFromPod(pod, clusterID)
  341. if err != nil {
  342. return nil, err
  343. }
  344. for _, c := range cs {
  345. containers[c.Key()] = true // captures any containers that existed for a time < a prometheus scrape interval. We currently charge 0 for this but should charge something.
  346. currentContainers[c.Key()] = *pod
  347. }
  348. }
  349. missingNodes := make(map[string]*costAnalyzerCloud.Node)
  350. missingContainers := make(map[string]*CostData)
  351. for key := range containers {
  352. if _, ok := containerNameCost[key]; ok {
  353. continue // because ordering is important for the allocation model (all PV's applied to the first), just dedupe if it's already been added.
  354. }
  355. // The _else_ case for this statement is the case in which the container has been
  356. // deleted so we have usage information but not request information. In that case,
  357. // we return partial data for CPU and RAM: only usage and not requests.
  358. if pod, ok := currentContainers[key]; ok {
  359. podName := pod.GetObjectMeta().GetName()
  360. ns := pod.GetObjectMeta().GetNamespace()
  361. nsLabels := namespaceLabelsMapping[ns+","+clusterID]
  362. podLabels := pod.GetObjectMeta().GetLabels()
  363. if podLabels == nil {
  364. podLabels = make(map[string]string)
  365. }
  366. for k, v := range nsLabels {
  367. if _, ok := podLabels[k]; !ok {
  368. podLabels[k] = v
  369. }
  370. }
  371. nsAnnotations := namespaceAnnotationsMapping[ns+","+clusterID]
  372. podAnnotations := pod.GetObjectMeta().GetAnnotations()
  373. if podAnnotations == nil {
  374. podAnnotations = make(map[string]string)
  375. }
  376. for k, v := range nsAnnotations {
  377. if _, ok := podAnnotations[k]; !ok {
  378. podAnnotations[k] = v
  379. }
  380. }
  381. nodeName := pod.Spec.NodeName
  382. var nodeData *costAnalyzerCloud.Node
  383. if _, ok := nodes[nodeName]; ok {
  384. nodeData = nodes[nodeName]
  385. }
  386. nsKey := ns + "," + clusterID
  387. var podDeployments []string
  388. if _, ok := podDeploymentsMapping[nsKey]; ok {
  389. if ds, ok := podDeploymentsMapping[nsKey][pod.GetObjectMeta().GetName()]; ok {
  390. podDeployments = ds
  391. } else {
  392. podDeployments = []string{}
  393. }
  394. }
  395. var podPVs []*PersistentVolumeClaimData
  396. podClaims := pod.Spec.Volumes
  397. for _, vol := range podClaims {
  398. if vol.PersistentVolumeClaim != nil {
  399. name := vol.PersistentVolumeClaim.ClaimName
  400. key := ns + "," + name + "," + clusterID
  401. if pvClaim, ok := pvClaimMapping[key]; ok {
  402. pvClaim.TimesClaimed++
  403. podPVs = append(podPVs, pvClaim)
  404. // Remove entry from potential unmounted pvs
  405. delete(unmountedPVs, key)
  406. }
  407. }
  408. }
  409. var podNetCosts []*util.Vector
  410. if usage, ok := networkUsageMap[ns+","+podName+","+clusterID]; ok {
  411. netCosts, err := GetNetworkCost(usage, cp)
  412. if err != nil {
  413. klog.V(4).Infof("Error pulling network costs: %s", err.Error())
  414. } else {
  415. podNetCosts = netCosts
  416. }
  417. }
  418. var podServices []string
  419. if _, ok := podServicesMapping[nsKey]; ok {
  420. if svcs, ok := podServicesMapping[nsKey][pod.GetObjectMeta().GetName()]; ok {
  421. podServices = svcs
  422. } else {
  423. podServices = []string{}
  424. }
  425. }
  426. for i, container := range pod.Spec.Containers {
  427. containerName := container.Name
  428. // recreate the key and look up data for this container
  429. newKey := NewContainerMetricFromValues(ns, podName, containerName, pod.Spec.NodeName, clusterID).Key()
  430. // k8s.io/apimachinery/pkg/api/resource/amount.go and
  431. // k8s.io/apimachinery/pkg/api/resource/quantity.go for
  432. // details on the "amount" API. See
  433. // https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/#resource-types
  434. // for the units of memory and CPU.
  435. ramRequestBytes := container.Resources.Requests.Memory().Value()
  436. // Because RAM (and CPU) information isn't coming from Prometheus, it won't
  437. // have a timestamp associated with it. We need to provide a timestamp,
  438. // otherwise the vector op that gets applied to take the max of usage
  439. // and request won't work properly and will only take into account
  440. // usage.
  441. RAMReqV := []*util.Vector{
  442. {
  443. Value: float64(ramRequestBytes),
  444. Timestamp: float64(time.Now().UTC().Unix()),
  445. },
  446. }
  447. // use millicores so we can convert to cores in a float64 format
  448. cpuRequestMilliCores := container.Resources.Requests.Cpu().MilliValue()
  449. CPUReqV := []*util.Vector{
  450. {
  451. Value: float64(cpuRequestMilliCores) / 1000,
  452. Timestamp: float64(time.Now().UTC().Unix()),
  453. },
  454. }
  455. RAMUsedV, ok := RAMUsedMap[newKey]
  456. if !ok {
  457. klog.V(4).Info("no RAM usage for " + newKey)
  458. RAMUsedV = []*util.Vector{{}}
  459. }
  460. GPUReqV, ok := GPUReqMap[newKey]
  461. if !ok {
  462. klog.V(4).Info("no GPU requests for " + newKey)
  463. GPUReqV = []*util.Vector{{}}
  464. }
  465. CPUUsedV, ok := CPUUsedMap[newKey]
  466. if !ok {
  467. klog.V(4).Info("no CPU usage for " + newKey)
  468. CPUUsedV = []*util.Vector{{}}
  469. }
  470. var pvReq []*PersistentVolumeClaimData
  471. var netReq []*util.Vector
  472. if i == 0 { // avoid duplicating by just assigning all claims to the first container.
  473. pvReq = podPVs
  474. netReq = podNetCosts
  475. }
  476. costs := &CostData{
  477. Name: containerName,
  478. PodName: podName,
  479. NodeName: nodeName,
  480. Namespace: ns,
  481. Deployments: podDeployments,
  482. Services: podServices,
  483. Daemonsets: getDaemonsetsOfPod(pod),
  484. Jobs: getJobsOfPod(pod),
  485. Statefulsets: getStatefulSetsOfPod(pod),
  486. NodeData: nodeData,
  487. RAMReq: RAMReqV,
  488. RAMUsed: RAMUsedV,
  489. CPUReq: CPUReqV,
  490. CPUUsed: CPUUsedV,
  491. GPUReq: GPUReqV,
  492. PVCData: pvReq,
  493. NetworkData: netReq,
  494. Annotations: podAnnotations,
  495. Labels: podLabels,
  496. NamespaceLabels: nsLabels,
  497. ClusterID: clusterID,
  498. ClusterName: cm.ClusterMap.NameFor(clusterID),
  499. }
  500. costs.CPUAllocation = getContainerAllocation(costs.CPUReq, costs.CPUUsed, "CPU")
  501. costs.RAMAllocation = getContainerAllocation(costs.RAMReq, costs.RAMUsed, "RAM")
  502. if filterNamespace == "" {
  503. containerNameCost[newKey] = costs
  504. } else if costs.Namespace == filterNamespace {
  505. containerNameCost[newKey] = costs
  506. }
  507. }
  508. } else {
  509. // The container has been deleted. Not all information is sent to prometheus via ksm, so fill out what we can without k8s api
  510. klog.V(4).Info("The container " + key + " has been deleted. Calculating allocation but resulting object will be missing data.")
  511. c, err := NewContainerMetricFromKey(key)
  512. if err != nil {
  513. return nil, err
  514. }
  515. // CPU and RAM requests are obtained from the Kubernetes API.
  516. // If this case has been reached, the Kubernetes API will not
  517. // have information about the pod because it no longer exists.
  518. //
  519. // The case where this matters is minimal, mainly in environments
  520. // with very short-lived pods that over-request resources.
  521. RAMReqV := []*util.Vector{{}}
  522. CPUReqV := []*util.Vector{{}}
  523. RAMUsedV, ok := RAMUsedMap[key]
  524. if !ok {
  525. klog.V(4).Info("no RAM usage for " + key)
  526. RAMUsedV = []*util.Vector{{}}
  527. }
  528. GPUReqV, ok := GPUReqMap[key]
  529. if !ok {
  530. klog.V(4).Info("no GPU requests for " + key)
  531. GPUReqV = []*util.Vector{{}}
  532. }
  533. CPUUsedV, ok := CPUUsedMap[key]
  534. if !ok {
  535. klog.V(4).Info("no CPU usage for " + key)
  536. CPUUsedV = []*util.Vector{{}}
  537. }
  538. node, ok := nodes[c.NodeName]
  539. if !ok {
  540. klog.V(4).Infof("Node \"%s\" has been deleted from Kubernetes. Query historical data to get it.", c.NodeName)
  541. if n, ok := missingNodes[c.NodeName]; ok {
  542. node = n
  543. } else {
  544. node = &costAnalyzerCloud.Node{}
  545. missingNodes[c.NodeName] = node
  546. }
  547. }
  548. namespacelabels, ok := namespaceLabelsMapping[c.Namespace+","+c.ClusterID]
  549. if !ok {
  550. klog.V(3).Infof("Missing data for namespace %s", c.Namespace)
  551. }
  552. namespaceAnnotations, ok := namespaceAnnotationsMapping[c.Namespace+","+c.ClusterID]
  553. if !ok {
  554. klog.V(3).Infof("Missing data for namespace %s", c.Namespace)
  555. }
  556. costs := &CostData{
  557. Name: c.ContainerName,
  558. PodName: c.PodName,
  559. NodeName: c.NodeName,
  560. NodeData: node,
  561. Namespace: c.Namespace,
  562. RAMReq: RAMReqV,
  563. RAMUsed: RAMUsedV,
  564. CPUReq: CPUReqV,
  565. CPUUsed: CPUUsedV,
  566. GPUReq: GPUReqV,
  567. Annotations: namespaceAnnotations,
  568. NamespaceLabels: namespacelabels,
  569. ClusterID: c.ClusterID,
  570. ClusterName: cm.ClusterMap.NameFor(c.ClusterID),
  571. }
  572. costs.CPUAllocation = getContainerAllocation(costs.CPUReq, costs.CPUUsed, "CPU")
  573. costs.RAMAllocation = getContainerAllocation(costs.RAMReq, costs.RAMUsed, "RAM")
  574. if filterNamespace == "" {
  575. containerNameCost[key] = costs
  576. missingContainers[key] = costs
  577. } else if costs.Namespace == filterNamespace {
  578. containerNameCost[key] = costs
  579. missingContainers[key] = costs
  580. }
  581. }
  582. }
  583. // Use unmounted pvs to create a mapping of "Unmounted-<Namespace>" containers
  584. // to pass along the cost data
  585. unmounted := findUnmountedPVCostData(cm.ClusterMap, unmountedPVs, namespaceLabelsMapping, namespaceAnnotationsMapping)
  586. for k, costs := range unmounted {
  587. klog.V(4).Infof("Unmounted PVs in Namespace/ClusterID: %s/%s", costs.Namespace, costs.ClusterID)
  588. if filterNamespace == "" {
  589. containerNameCost[k] = costs
  590. } else if costs.Namespace == filterNamespace {
  591. containerNameCost[k] = costs
  592. }
  593. }
  594. err = findDeletedNodeInfo(cli, missingNodes, window, "")
  595. if err != nil {
  596. klog.V(1).Infof("Error fetching historical node data: %s", err.Error())
  597. }
  598. err = findDeletedPodInfo(cli, missingContainers, window)
  599. if err != nil {
  600. klog.V(1).Infof("Error fetching historical pod data: %s", err.Error())
  601. }
  602. return containerNameCost, err
  603. }
  604. func findUnmountedPVCostData(clusterMap clusters.ClusterMap, unmountedPVs map[string][]*PersistentVolumeClaimData, namespaceLabelsMapping map[string]map[string]string, namespaceAnnotationsMapping map[string]map[string]string) map[string]*CostData {
  605. costs := make(map[string]*CostData)
  606. if len(unmountedPVs) == 0 {
  607. return costs
  608. }
  609. for k, pv := range unmountedPVs {
  610. keyParts := strings.Split(k, ",")
  611. if len(keyParts) != 3 {
  612. klog.V(1).Infof("Unmounted PV used key with incorrect parts: %s", k)
  613. continue
  614. }
  615. ns, _, clusterID := keyParts[0], keyParts[1], keyParts[2]
  616. namespacelabels, ok := namespaceLabelsMapping[ns+","+clusterID]
  617. if !ok {
  618. klog.V(3).Infof("Missing data for namespace %s", ns)
  619. }
  620. namespaceAnnotations, ok := namespaceAnnotationsMapping[ns+","+clusterID]
  621. if !ok {
  622. klog.V(3).Infof("Missing data for namespace %s", ns)
  623. }
  624. // Should be a unique "Unmounted" cost data type
  625. name := "unmounted-pvs"
  626. metric := NewContainerMetricFromValues(ns, name, name, "", clusterID)
  627. key := metric.Key()
  628. if costData, ok := costs[key]; !ok {
  629. costs[key] = &CostData{
  630. Name: name,
  631. PodName: name,
  632. NodeName: "",
  633. Annotations: namespaceAnnotations,
  634. Namespace: ns,
  635. NamespaceLabels: namespacelabels,
  636. Labels: namespacelabels,
  637. ClusterID: clusterID,
  638. ClusterName: clusterMap.NameFor(clusterID),
  639. PVCData: pv,
  640. }
  641. } else {
  642. costData.PVCData = append(costData.PVCData, pv...)
  643. }
  644. }
  645. return costs
  646. }
  647. func findDeletedPodInfo(cli prometheusClient.Client, missingContainers map[string]*CostData, window string) error {
  648. if len(missingContainers) > 0 {
  649. queryHistoricalPodLabels := fmt.Sprintf(`kube_pod_labels{}[%s]`, window)
  650. podLabelsResult, _, err := prom.NewContext(cli).QuerySync(queryHistoricalPodLabels)
  651. if err != nil {
  652. log.Errorf("failed to parse historical pod labels: %s", err.Error())
  653. }
  654. podLabels := make(map[string]map[string]string)
  655. if podLabelsResult != nil {
  656. podLabels, err = parsePodLabels(podLabelsResult)
  657. if err != nil {
  658. log.Errorf("failed to parse historical pod labels: %s", err.Error())
  659. }
  660. }
  661. for key, costData := range missingContainers {
  662. cm, _ := NewContainerMetricFromKey(key)
  663. labels, ok := podLabels[cm.PodName]
  664. if !ok {
  665. labels = make(map[string]string)
  666. }
  667. for k, v := range costData.NamespaceLabels {
  668. labels[k] = v
  669. }
  670. costData.Labels = labels
  671. }
  672. }
  673. return nil
  674. }
  675. func findDeletedNodeInfo(cli prometheusClient.Client, missingNodes map[string]*costAnalyzerCloud.Node, window, offset string) error {
  676. if len(missingNodes) > 0 {
  677. defer measureTime(time.Now(), profileThreshold, "Finding Deleted Node Info")
  678. offsetStr := ""
  679. if offset != "" {
  680. offsetStr = fmt.Sprintf("offset %s", offset)
  681. }
  682. queryHistoricalCPUCost := fmt.Sprintf(`avg(avg_over_time(node_cpu_hourly_cost[%s] %s)) by (node, instance, cluster_id)`, window, offsetStr)
  683. queryHistoricalRAMCost := fmt.Sprintf(`avg(avg_over_time(node_ram_hourly_cost[%s] %s)) by (node, instance, cluster_id)`, window, offsetStr)
  684. queryHistoricalGPUCost := fmt.Sprintf(`avg(avg_over_time(node_gpu_hourly_cost[%s] %s)) by (node, instance, cluster_id)`, window, offsetStr)
  685. ctx := prom.NewContext(cli)
  686. cpuCostResCh := ctx.Query(queryHistoricalCPUCost)
  687. ramCostResCh := ctx.Query(queryHistoricalRAMCost)
  688. gpuCostResCh := ctx.Query(queryHistoricalGPUCost)
  689. cpuCostRes, _ := cpuCostResCh.Await()
  690. ramCostRes, _ := ramCostResCh.Await()
  691. gpuCostRes, _ := gpuCostResCh.Await()
  692. if ctx.HasErrors() {
  693. return ctx.ErrorCollection()
  694. }
  695. cpuCosts, err := getCost(cpuCostRes)
  696. if err != nil {
  697. return err
  698. }
  699. ramCosts, err := getCost(ramCostRes)
  700. if err != nil {
  701. return err
  702. }
  703. gpuCosts, err := getCost(gpuCostRes)
  704. if err != nil {
  705. return err
  706. }
  707. if len(cpuCosts) == 0 {
  708. klog.V(1).Infof("Kubecost prometheus metrics not currently available. Ingest this server's /metrics endpoint to get that data.")
  709. }
  710. for node, costv := range cpuCosts {
  711. if _, ok := missingNodes[node]; ok {
  712. missingNodes[node].VCPUCost = fmt.Sprintf("%f", costv[0].Value)
  713. } else {
  714. log.DedupedWarningf(5, "Node `%s` in prometheus but not k8s api", node)
  715. }
  716. }
  717. for node, costv := range ramCosts {
  718. if _, ok := missingNodes[node]; ok {
  719. missingNodes[node].RAMCost = fmt.Sprintf("%f", costv[0].Value)
  720. }
  721. }
  722. for node, costv := range gpuCosts {
  723. if _, ok := missingNodes[node]; ok {
  724. missingNodes[node].GPUCost = fmt.Sprintf("%f", costv[0].Value)
  725. }
  726. }
  727. }
  728. return nil
  729. }
  730. func getContainerAllocation(req []*util.Vector, used []*util.Vector, allocationType string) []*util.Vector {
  731. // The result of the normalize operation will be a new []*util.Vector to replace the requests
  732. allocationOp := func(r *util.Vector, x *float64, y *float64) bool {
  733. if x != nil && y != nil {
  734. x1 := *x
  735. if math.IsNaN(x1) {
  736. klog.V(1).Infof("[Warning] NaN value found during %s allocation calculation for requests.", allocationType)
  737. x1 = 0.0
  738. }
  739. y1 := *y
  740. if math.IsNaN(y1) {
  741. klog.V(1).Infof("[Warning] NaN value found during %s allocation calculation for used.", allocationType)
  742. y1 = 0.0
  743. }
  744. r.Value = math.Max(x1, y1)
  745. } else if x != nil {
  746. r.Value = *x
  747. } else if y != nil {
  748. r.Value = *y
  749. }
  750. return true
  751. }
  752. return util.ApplyVectorOp(req, used, allocationOp)
  753. }
  754. func addPVData(cache clustercache.ClusterCache, pvClaimMapping map[string]*PersistentVolumeClaimData, cloud costAnalyzerCloud.Provider) error {
  755. cfg, err := cloud.GetConfig()
  756. if err != nil {
  757. return err
  758. }
  759. // Pull a region from the first node
  760. var defaultRegion string
  761. nodeList := cache.GetAllNodes()
  762. if len(nodeList) > 0 {
  763. defaultRegion, _ = util.GetRegion(nodeList[0].Labels)
  764. }
  765. storageClasses := cache.GetAllStorageClasses()
  766. storageClassMap := make(map[string]map[string]string)
  767. for _, storageClass := range storageClasses {
  768. params := storageClass.Parameters
  769. storageClassMap[storageClass.ObjectMeta.Name] = params
  770. if storageClass.GetAnnotations()["storageclass.kubernetes.io/is-default-class"] == "true" || storageClass.GetAnnotations()["storageclass.beta.kubernetes.io/is-default-class"] == "true" {
  771. storageClassMap["default"] = params
  772. storageClassMap[""] = params
  773. }
  774. }
  775. pvs := cache.GetAllPersistentVolumes()
  776. pvMap := make(map[string]*costAnalyzerCloud.PV)
  777. for _, pv := range pvs {
  778. parameters, ok := storageClassMap[pv.Spec.StorageClassName]
  779. if !ok {
  780. klog.V(4).Infof("Unable to find parameters for storage class \"%s\". Does pv \"%s\" have a storageClassName?", pv.Spec.StorageClassName, pv.Name)
  781. }
  782. var region string
  783. if r, ok := util.GetRegion(pv.Labels); ok {
  784. region = r
  785. } else {
  786. region = defaultRegion
  787. }
  788. cacPv := &costAnalyzerCloud.PV{
  789. Class: pv.Spec.StorageClassName,
  790. Region: region,
  791. Parameters: parameters,
  792. }
  793. err := GetPVCost(cacPv, pv, cloud, region)
  794. if err != nil {
  795. return err
  796. }
  797. pvMap[pv.Name] = cacPv
  798. }
  799. for _, pvc := range pvClaimMapping {
  800. if vol, ok := pvMap[pvc.VolumeName]; ok {
  801. pvc.Volume = vol
  802. } else {
  803. klog.V(4).Infof("PV not found, using default")
  804. pvc.Volume = &costAnalyzerCloud.PV{
  805. Cost: cfg.Storage,
  806. }
  807. }
  808. }
  809. return nil
  810. }
  811. func GetPVCost(pv *costAnalyzerCloud.PV, kpv *v1.PersistentVolume, cp costAnalyzerCloud.Provider, defaultRegion string) error {
  812. cfg, err := cp.GetConfig()
  813. if err != nil {
  814. return err
  815. }
  816. key := cp.GetPVKey(kpv, pv.Parameters, defaultRegion)
  817. pvWithCost, err := cp.PVPricing(key)
  818. if err != nil {
  819. pv.Cost = cfg.Storage
  820. return err
  821. }
  822. if pvWithCost == nil || pvWithCost.Cost == "" {
  823. pv.Cost = cfg.Storage
  824. return nil // set default cost
  825. }
  826. pv.Cost = pvWithCost.Cost
  827. pv.ProviderID = key.ID()
  828. return nil
  829. }
  830. func (cm *CostModel) GetPricingSourceCounts() (*costAnalyzerCloud.PricingMatchMetadata, error) {
  831. if cm.pricingMetadata != nil {
  832. return cm.pricingMetadata, nil
  833. } else {
  834. return nil, fmt.Errorf("Node costs not yet calculated")
  835. }
  836. }
  837. func (cm *CostModel) GetNodeCost(cp costAnalyzerCloud.Provider) (map[string]*costAnalyzerCloud.Node, error) {
  838. cfg, err := cp.GetConfig()
  839. if err != nil {
  840. return nil, err
  841. }
  842. nodeList := cm.Cache.GetAllNodes()
  843. nodes := make(map[string]*costAnalyzerCloud.Node)
  844. pmd := &costAnalyzerCloud.PricingMatchMetadata{
  845. TotalNodes: 0,
  846. PricingTypeCounts: make(map[costAnalyzerCloud.PricingType]int),
  847. }
  848. for _, n := range nodeList {
  849. name := n.GetObjectMeta().GetName()
  850. nodeLabels := n.GetObjectMeta().GetLabels()
  851. nodeLabels["providerID"] = n.Spec.ProviderID
  852. pmd.TotalNodes++
  853. cnode, err := cp.NodePricing(cp.GetKey(nodeLabels, n))
  854. if err != nil {
  855. klog.Infof("Error getting node pricing. Error: %s", err.Error())
  856. if cnode != nil {
  857. nodes[name] = cnode
  858. continue
  859. } else {
  860. cnode = &costAnalyzerCloud.Node{
  861. VCPUCost: cfg.CPU,
  862. RAMCost: cfg.RAM,
  863. }
  864. }
  865. }
  866. if _, ok := pmd.PricingTypeCounts[cnode.PricingType]; ok {
  867. pmd.PricingTypeCounts[cnode.PricingType]++
  868. } else {
  869. pmd.PricingTypeCounts[cnode.PricingType] = 1
  870. }
  871. newCnode := *cnode
  872. if newCnode.InstanceType == "" {
  873. it, _ := util.GetInstanceType(n.Labels)
  874. newCnode.InstanceType = it
  875. }
  876. if newCnode.Region == "" {
  877. region, _ := util.GetRegion(n.Labels)
  878. newCnode.Region = region
  879. }
  880. newCnode.ProviderID = n.Spec.ProviderID
  881. var cpu float64
  882. if newCnode.VCPU == "" {
  883. cpu = float64(n.Status.Capacity.Cpu().Value())
  884. newCnode.VCPU = n.Status.Capacity.Cpu().String()
  885. } else {
  886. cpu, err = strconv.ParseFloat(newCnode.VCPU, 64)
  887. if err != nil {
  888. klog.V(1).Infof("[Warning] parsing VCPU value: \"%s\" as float64", newCnode.VCPU)
  889. }
  890. }
  891. if math.IsNaN(cpu) {
  892. klog.V(1).Infof("[Warning] cpu parsed as NaN. Setting to 0.")
  893. cpu = 0
  894. }
  895. var ram float64
  896. if newCnode.RAM == "" {
  897. newCnode.RAM = n.Status.Capacity.Memory().String()
  898. }
  899. ram = float64(n.Status.Capacity.Memory().Value())
  900. if math.IsNaN(ram) {
  901. klog.V(1).Infof("[Warning] ram parsed as NaN. Setting to 0.")
  902. ram = 0
  903. }
  904. newCnode.RAMBytes = fmt.Sprintf("%f", ram)
  905. // Azure does not seem to provide a GPU count in its pricing API. GKE supports attaching multiple GPUs
  906. // So the k8s api will often report more accurate results for GPU count under status > capacity > nvidia.com/gpu than the cloud providers billing data
  907. // not all providers are guaranteed to use this, so don't overwrite a Provider assignment if we can't find something under that capacity exists
  908. gpuc := 0.0
  909. q, ok := n.Status.Capacity["nvidia.com/gpu"]
  910. if ok {
  911. gpuCount := q.Value()
  912. if gpuCount != 0 {
  913. newCnode.GPU = fmt.Sprintf("%d", q.Value())
  914. gpuc = float64(gpuCount)
  915. }
  916. } else {
  917. gpuc, err = strconv.ParseFloat(newCnode.GPU, 64)
  918. if err != nil {
  919. gpuc = 0.0
  920. }
  921. }
  922. if math.IsNaN(gpuc) {
  923. klog.V(1).Infof("[Warning] gpu count parsed as NaN. Setting to 0.")
  924. gpuc = 0.0
  925. }
  926. if newCnode.GPU != "" && newCnode.GPUCost == "" {
  927. // We couldn't find a gpu cost, so fix cpu and ram, then accordingly
  928. klog.V(4).Infof("GPU without cost found for %s, calculating...", cp.GetKey(nodeLabels, n).Features())
  929. defaultCPU, err := strconv.ParseFloat(cfg.CPU, 64)
  930. if err != nil {
  931. klog.V(3).Infof("Could not parse default cpu price")
  932. defaultCPU = 0
  933. }
  934. if math.IsNaN(defaultCPU) {
  935. klog.V(1).Infof("[Warning] defaultCPU parsed as NaN. Setting to 0.")
  936. defaultCPU = 0
  937. }
  938. defaultRAM, err := strconv.ParseFloat(cfg.RAM, 64)
  939. if err != nil {
  940. klog.V(3).Infof("Could not parse default ram price")
  941. defaultRAM = 0
  942. }
  943. if math.IsNaN(defaultRAM) {
  944. klog.V(1).Infof("[Warning] defaultRAM parsed as NaN. Setting to 0.")
  945. defaultRAM = 0
  946. }
  947. defaultGPU, err := strconv.ParseFloat(cfg.GPU, 64)
  948. if err != nil {
  949. klog.V(3).Infof("Could not parse default gpu price")
  950. defaultGPU = 0
  951. }
  952. if math.IsNaN(defaultGPU) {
  953. klog.V(1).Infof("[Warning] defaultGPU parsed as NaN. Setting to 0.")
  954. defaultGPU = 0
  955. }
  956. cpuToRAMRatio := defaultCPU / defaultRAM
  957. if math.IsNaN(cpuToRAMRatio) {
  958. klog.V(1).Infof("[Warning] cpuToRAMRatio[defaultCPU: %f / defaultRam: %f] is NaN. Setting to 0.", defaultCPU, defaultRAM)
  959. cpuToRAMRatio = 0
  960. }
  961. gpuToRAMRatio := defaultGPU / defaultRAM
  962. if math.IsNaN(gpuToRAMRatio) {
  963. klog.V(1).Infof("[Warning] gpuToRAMRatio is NaN. Setting to 0.")
  964. gpuToRAMRatio = 0
  965. }
  966. ramGB := ram / 1024 / 1024 / 1024
  967. if math.IsNaN(ramGB) {
  968. klog.V(1).Infof("[Warning] ramGB is NaN. Setting to 0.")
  969. ramGB = 0
  970. }
  971. ramMultiple := gpuc*gpuToRAMRatio + cpu*cpuToRAMRatio + ramGB
  972. if math.IsNaN(ramMultiple) {
  973. klog.V(1).Infof("[Warning] ramMultiple is NaN. Setting to 0.")
  974. ramMultiple = 0
  975. }
  976. var nodePrice float64
  977. if newCnode.Cost != "" {
  978. nodePrice, err = strconv.ParseFloat(newCnode.Cost, 64)
  979. if err != nil {
  980. klog.V(3).Infof("Could not parse total node price")
  981. return nil, err
  982. }
  983. } else {
  984. nodePrice, err = strconv.ParseFloat(newCnode.VCPUCost, 64) // all the price was allocated the the CPU
  985. if err != nil {
  986. klog.V(3).Infof("Could not parse node vcpu price")
  987. return nil, err
  988. }
  989. }
  990. if math.IsNaN(nodePrice) {
  991. klog.V(1).Infof("[Warning] nodePrice parsed as NaN. Setting to 0.")
  992. nodePrice = 0
  993. }
  994. ramPrice := (nodePrice / ramMultiple)
  995. if math.IsNaN(ramPrice) {
  996. klog.V(1).Infof("[Warning] ramPrice[nodePrice: %f / ramMultiple: %f] parsed as NaN. Setting to 0.", nodePrice, ramMultiple)
  997. ramPrice = 0
  998. }
  999. cpuPrice := ramPrice * cpuToRAMRatio
  1000. gpuPrice := ramPrice * gpuToRAMRatio
  1001. newCnode.VCPUCost = fmt.Sprintf("%f", cpuPrice)
  1002. newCnode.RAMCost = fmt.Sprintf("%f", ramPrice)
  1003. newCnode.RAMBytes = fmt.Sprintf("%f", ram)
  1004. newCnode.GPUCost = fmt.Sprintf("%f", gpuPrice)
  1005. } else if newCnode.RAMCost == "" {
  1006. // We couldn't find a ramcost, so fix cpu and allocate ram accordingly
  1007. klog.V(4).Infof("No RAM cost found for %s, calculating...", cp.GetKey(nodeLabels, n).Features())
  1008. defaultCPU, err := strconv.ParseFloat(cfg.CPU, 64)
  1009. if err != nil {
  1010. klog.V(3).Infof("Could not parse default cpu price")
  1011. defaultCPU = 0
  1012. }
  1013. if math.IsNaN(defaultCPU) {
  1014. klog.V(1).Infof("[Warning] defaultCPU parsed as NaN. Setting to 0.")
  1015. defaultCPU = 0
  1016. }
  1017. defaultRAM, err := strconv.ParseFloat(cfg.RAM, 64)
  1018. if err != nil {
  1019. klog.V(3).Infof("Could not parse default ram price")
  1020. defaultRAM = 0
  1021. }
  1022. if math.IsNaN(defaultRAM) {
  1023. klog.V(1).Infof("[Warning] defaultRAM parsed as NaN. Setting to 0.")
  1024. defaultRAM = 0
  1025. }
  1026. cpuToRAMRatio := defaultCPU / defaultRAM
  1027. if math.IsNaN(cpuToRAMRatio) {
  1028. klog.V(1).Infof("[Warning] cpuToRAMRatio[defaultCPU: %f / defaultRam: %f] is NaN. Setting to 0.", defaultCPU, defaultRAM)
  1029. cpuToRAMRatio = 0
  1030. }
  1031. ramGB := ram / 1024 / 1024 / 1024
  1032. if math.IsNaN(ramGB) {
  1033. klog.V(1).Infof("[Warning] ramGB is NaN. Setting to 0.")
  1034. ramGB = 0
  1035. }
  1036. ramMultiple := cpu*cpuToRAMRatio + ramGB
  1037. if math.IsNaN(ramMultiple) {
  1038. klog.V(1).Infof("[Warning] ramMultiple is NaN. Setting to 0.")
  1039. ramMultiple = 0
  1040. }
  1041. var nodePrice float64
  1042. if newCnode.Cost != "" {
  1043. nodePrice, err = strconv.ParseFloat(newCnode.Cost, 64)
  1044. if err != nil {
  1045. klog.V(3).Infof("Could not parse total node price")
  1046. return nil, err
  1047. }
  1048. } else {
  1049. nodePrice, err = strconv.ParseFloat(newCnode.VCPUCost, 64) // all the price was allocated the the CPU
  1050. if err != nil {
  1051. klog.V(3).Infof("Could not parse node vcpu price")
  1052. return nil, err
  1053. }
  1054. }
  1055. if math.IsNaN(nodePrice) {
  1056. klog.V(1).Infof("[Warning] nodePrice parsed as NaN. Setting to 0.")
  1057. nodePrice = 0
  1058. }
  1059. ramPrice := (nodePrice / ramMultiple)
  1060. if math.IsNaN(ramPrice) {
  1061. klog.V(1).Infof("[Warning] ramPrice[nodePrice: %f / ramMultiple: %f] parsed as NaN. Setting to 0.", nodePrice, ramMultiple)
  1062. ramPrice = 0
  1063. }
  1064. cpuPrice := ramPrice * cpuToRAMRatio
  1065. if defaultRAM != 0 {
  1066. newCnode.VCPUCost = fmt.Sprintf("%f", cpuPrice)
  1067. newCnode.RAMCost = fmt.Sprintf("%f", ramPrice)
  1068. } else { // just assign the full price to CPU
  1069. if cpu != 0 {
  1070. newCnode.VCPUCost = fmt.Sprintf("%f", nodePrice/cpu)
  1071. } else {
  1072. newCnode.VCPUCost = fmt.Sprintf("%f", nodePrice)
  1073. }
  1074. }
  1075. newCnode.RAMBytes = fmt.Sprintf("%f", ram)
  1076. klog.V(4).Infof("Computed \"%s\" RAM Cost := %v", name, newCnode.RAMCost)
  1077. }
  1078. nodes[name] = &newCnode
  1079. }
  1080. cm.pricingMetadata = pmd
  1081. cp.ApplyReservedInstancePricing(nodes)
  1082. return nodes, nil
  1083. }
  1084. // TODO: drop some logs
  1085. func (cm *CostModel) GetLBCost(cp costAnalyzerCloud.Provider) (map[string]*costAnalyzerCloud.LoadBalancer, error) {
  1086. // for fetching prices from cloud provider
  1087. // cfg, err := cp.GetConfig()
  1088. // if err != nil {
  1089. // return nil, err
  1090. // }
  1091. servicesList := cm.Cache.GetAllServices()
  1092. loadBalancerMap := make(map[string]*costAnalyzerCloud.LoadBalancer)
  1093. for _, service := range servicesList {
  1094. namespace := service.GetObjectMeta().GetNamespace()
  1095. name := service.GetObjectMeta().GetName()
  1096. key := namespace + "," + name // + "," + clusterID?
  1097. if service.Spec.Type == "LoadBalancer" {
  1098. loadBalancer, err := cp.LoadBalancerPricing()
  1099. if err != nil {
  1100. return nil, err
  1101. }
  1102. newLoadBalancer := *loadBalancer
  1103. for _, loadBalancerIngress := range service.Status.LoadBalancer.Ingress {
  1104. newLoadBalancer.IngressIPAddresses = append(newLoadBalancer.IngressIPAddresses, loadBalancerIngress.IP)
  1105. }
  1106. loadBalancerMap[key] = &newLoadBalancer
  1107. }
  1108. }
  1109. return loadBalancerMap, nil
  1110. }
  1111. func getPodServices(cache clustercache.ClusterCache, podList []*v1.Pod, clusterID string) (map[string]map[string][]string, error) {
  1112. servicesList := cache.GetAllServices()
  1113. podServicesMapping := make(map[string]map[string][]string)
  1114. for _, service := range servicesList {
  1115. namespace := service.GetObjectMeta().GetNamespace()
  1116. name := service.GetObjectMeta().GetName()
  1117. key := namespace + "," + clusterID
  1118. if _, ok := podServicesMapping[key]; !ok {
  1119. podServicesMapping[key] = make(map[string][]string)
  1120. }
  1121. s := labels.Nothing()
  1122. if service.Spec.Selector != nil && len(service.Spec.Selector) > 0 {
  1123. s = labels.Set(service.Spec.Selector).AsSelectorPreValidated()
  1124. }
  1125. for _, pod := range podList {
  1126. labelSet := labels.Set(pod.GetObjectMeta().GetLabels())
  1127. if s.Matches(labelSet) && pod.GetObjectMeta().GetNamespace() == namespace {
  1128. services, ok := podServicesMapping[key][pod.GetObjectMeta().GetName()]
  1129. if ok {
  1130. podServicesMapping[key][pod.GetObjectMeta().GetName()] = append(services, name)
  1131. } else {
  1132. podServicesMapping[key][pod.GetObjectMeta().GetName()] = []string{name}
  1133. }
  1134. }
  1135. }
  1136. }
  1137. return podServicesMapping, nil
  1138. }
  1139. func getPodStatefulsets(cache clustercache.ClusterCache, podList []*v1.Pod, clusterID string) (map[string]map[string][]string, error) {
  1140. ssList := cache.GetAllStatefulSets()
  1141. podSSMapping := make(map[string]map[string][]string) // namespace: podName: [deploymentNames]
  1142. for _, ss := range ssList {
  1143. namespace := ss.GetObjectMeta().GetNamespace()
  1144. name := ss.GetObjectMeta().GetName()
  1145. key := namespace + "," + clusterID
  1146. if _, ok := podSSMapping[key]; !ok {
  1147. podSSMapping[key] = make(map[string][]string)
  1148. }
  1149. s, err := metav1.LabelSelectorAsSelector(ss.Spec.Selector)
  1150. if err != nil {
  1151. klog.V(2).Infof("Error doing deployment label conversion: " + err.Error())
  1152. }
  1153. for _, pod := range podList {
  1154. labelSet := labels.Set(pod.GetObjectMeta().GetLabels())
  1155. if s.Matches(labelSet) && pod.GetObjectMeta().GetNamespace() == namespace {
  1156. sss, ok := podSSMapping[key][pod.GetObjectMeta().GetName()]
  1157. if ok {
  1158. podSSMapping[key][pod.GetObjectMeta().GetName()] = append(sss, name)
  1159. } else {
  1160. podSSMapping[key][pod.GetObjectMeta().GetName()] = []string{name}
  1161. }
  1162. }
  1163. }
  1164. }
  1165. return podSSMapping, nil
  1166. }
  1167. func getPodDeployments(cache clustercache.ClusterCache, podList []*v1.Pod, clusterID string) (map[string]map[string][]string, error) {
  1168. deploymentsList := cache.GetAllDeployments()
  1169. podDeploymentsMapping := make(map[string]map[string][]string) // namespace: podName: [deploymentNames]
  1170. for _, deployment := range deploymentsList {
  1171. namespace := deployment.GetObjectMeta().GetNamespace()
  1172. name := deployment.GetObjectMeta().GetName()
  1173. key := namespace + "," + clusterID
  1174. if _, ok := podDeploymentsMapping[key]; !ok {
  1175. podDeploymentsMapping[key] = make(map[string][]string)
  1176. }
  1177. s, err := metav1.LabelSelectorAsSelector(deployment.Spec.Selector)
  1178. if err != nil {
  1179. klog.V(2).Infof("Error doing deployment label conversion: " + err.Error())
  1180. }
  1181. for _, pod := range podList {
  1182. labelSet := labels.Set(pod.GetObjectMeta().GetLabels())
  1183. if s.Matches(labelSet) && pod.GetObjectMeta().GetNamespace() == namespace {
  1184. deployments, ok := podDeploymentsMapping[key][pod.GetObjectMeta().GetName()]
  1185. if ok {
  1186. podDeploymentsMapping[key][pod.GetObjectMeta().GetName()] = append(deployments, name)
  1187. } else {
  1188. podDeploymentsMapping[key][pod.GetObjectMeta().GetName()] = []string{name}
  1189. }
  1190. }
  1191. }
  1192. }
  1193. return podDeploymentsMapping, nil
  1194. }
  1195. func getPodDeploymentsWithMetrics(deploymentLabels map[string]map[string]string, podLabels map[string]map[string]string) (map[string]map[string][]string, error) {
  1196. podDeploymentsMapping := make(map[string]map[string][]string)
  1197. for depKey, depLabels := range deploymentLabels {
  1198. kt, err := NewKeyTuple(depKey)
  1199. if err != nil {
  1200. continue
  1201. }
  1202. namespace := kt.Namespace()
  1203. name := kt.Key()
  1204. clusterID := kt.ClusterID()
  1205. key := namespace + "," + clusterID
  1206. if _, ok := podDeploymentsMapping[key]; !ok {
  1207. podDeploymentsMapping[key] = make(map[string][]string)
  1208. }
  1209. s := labels.Set(depLabels).AsSelectorPreValidated()
  1210. for podKey, pLabels := range podLabels {
  1211. pkey, err := NewKeyTuple(podKey)
  1212. if err != nil {
  1213. continue
  1214. }
  1215. podNamespace := pkey.Namespace()
  1216. podName := pkey.Key()
  1217. podClusterID := pkey.ClusterID()
  1218. labelSet := labels.Set(pLabels)
  1219. if s.Matches(labelSet) && podNamespace == namespace && podClusterID == clusterID {
  1220. deployments, ok := podDeploymentsMapping[key][podName]
  1221. if ok {
  1222. podDeploymentsMapping[key][podName] = append(deployments, name)
  1223. } else {
  1224. podDeploymentsMapping[key][podName] = []string{name}
  1225. }
  1226. }
  1227. }
  1228. }
  1229. // Remove any duplicate data created by metric names
  1230. pruneDuplicateData(podDeploymentsMapping)
  1231. return podDeploymentsMapping, nil
  1232. }
  1233. func getPodServicesWithMetrics(serviceLabels map[string]map[string]string, podLabels map[string]map[string]string) (map[string]map[string][]string, error) {
  1234. podServicesMapping := make(map[string]map[string][]string)
  1235. for servKey, servLabels := range serviceLabels {
  1236. kt, err := NewKeyTuple(servKey)
  1237. if err != nil {
  1238. continue
  1239. }
  1240. namespace := kt.Namespace()
  1241. name := kt.Key()
  1242. clusterID := kt.ClusterID()
  1243. key := namespace + "," + clusterID
  1244. if _, ok := podServicesMapping[key]; !ok {
  1245. podServicesMapping[key] = make(map[string][]string)
  1246. }
  1247. s := labels.Nothing()
  1248. if servLabels != nil && len(servLabels) > 0 {
  1249. s = labels.Set(servLabels).AsSelectorPreValidated()
  1250. }
  1251. for podKey, pLabels := range podLabels {
  1252. pkey, err := NewKeyTuple(podKey)
  1253. if err != nil {
  1254. continue
  1255. }
  1256. podNamespace := pkey.Namespace()
  1257. podName := pkey.Key()
  1258. podClusterID := pkey.ClusterID()
  1259. labelSet := labels.Set(pLabels)
  1260. if s.Matches(labelSet) && podNamespace == namespace && podClusterID == clusterID {
  1261. services, ok := podServicesMapping[key][podName]
  1262. if ok {
  1263. podServicesMapping[key][podName] = append(services, name)
  1264. } else {
  1265. podServicesMapping[key][podName] = []string{name}
  1266. }
  1267. }
  1268. }
  1269. }
  1270. // Remove any duplicate data created by metric names
  1271. pruneDuplicateData(podServicesMapping)
  1272. return podServicesMapping, nil
  1273. }
  1274. // This method alleviates an issue with metrics that used a '_' to replace '-' in deployment
  1275. // and service names. To avoid counting these as multiple deployments/services, we'll remove
  1276. // the '_' version. Not optimal, but takes care of the issue
  1277. func pruneDuplicateData(data map[string]map[string][]string) {
  1278. for _, podMap := range data {
  1279. for podName, values := range podMap {
  1280. podMap[podName] = pruneDuplicates(values)
  1281. }
  1282. }
  1283. }
  1284. // Determine if there is an underscore in the value of a slice. If so, replace _ with -, and then
  1285. // check to see if the result exists in the slice. If both are true, then we DO NOT include that
  1286. // original value in the new slice.
  1287. func pruneDuplicates(s []string) []string {
  1288. m := sliceToSet(s)
  1289. for _, v := range s {
  1290. if strings.Contains(v, "_") {
  1291. name := strings.Replace(v, "_", "-", -1)
  1292. if !m[name] {
  1293. m[name] = true
  1294. }
  1295. delete(m, v)
  1296. }
  1297. }
  1298. return setToSlice(m)
  1299. }
  1300. // Creates a map[string]bool containing the slice values as keys
  1301. func sliceToSet(s []string) map[string]bool {
  1302. m := make(map[string]bool)
  1303. for _, v := range s {
  1304. m[v] = true
  1305. }
  1306. return m
  1307. }
  1308. func setToSlice(m map[string]bool) []string {
  1309. var result []string
  1310. for k := range m {
  1311. result = append(result, k)
  1312. }
  1313. return result
  1314. }
  1315. func costDataPassesFilters(cm clusters.ClusterMap, costs *CostData, namespace string, cluster string) bool {
  1316. passesNamespace := namespace == "" || costs.Namespace == namespace
  1317. passesCluster := cluster == "" || costs.ClusterID == cluster || costs.ClusterName == cluster
  1318. return passesNamespace && passesCluster
  1319. }
  1320. // Finds the a closest multiple less than value
  1321. func floorMultiple(value int64, multiple int64) int64 {
  1322. return (value / multiple) * multiple
  1323. }
  1324. // Attempt to create a key for the request. Reduce the times to minutes in order to more easily group requests based on
  1325. // real time ranges. If for any reason, the key generation fails, return a uuid to ensure uniqueness.
  1326. func requestKeyFor(window kubecost.Window, resolution time.Duration, filterNamespace string, filterCluster string, remoteEnabled bool) string {
  1327. keyLayout := "2006-01-02T15:04Z"
  1328. // We "snap" start time and duration to their closest 5 min multiple less than itself, by
  1329. // applying a snapped duration to a snapped start time.
  1330. durMins := int64(window.Minutes())
  1331. durMins = floorMultiple(durMins, 5)
  1332. sMins := int64(window.Start().Minute())
  1333. sOffset := sMins - floorMultiple(sMins, 5)
  1334. sTime := window.Start().Add(-time.Duration(sOffset) * time.Minute)
  1335. eTime := window.Start().Add(time.Duration(durMins) * time.Minute)
  1336. startKey := sTime.Format(keyLayout)
  1337. endKey := eTime.Format(keyLayout)
  1338. return fmt.Sprintf("%s,%s,%s,%s,%s,%t", startKey, endKey, resolution.String(), filterNamespace, filterCluster, remoteEnabled)
  1339. }
  1340. // ComputeCostDataRange executes a range query for cost data.
  1341. // Note that "offset" represents the time between the function call and "endString", and is also passed for convenience
  1342. func (cm *CostModel) ComputeCostDataRange(cli prometheusClient.Client, cp costAnalyzerCloud.Provider, window kubecost.Window, resolution time.Duration, filterNamespace string, filterCluster string, remoteEnabled bool) (map[string]*CostData, error) {
  1343. // Create a request key for request grouping. This key will be used to represent the cost-model result
  1344. // for the specific inputs to prevent multiple queries for identical data.
  1345. key := requestKeyFor(window, resolution, filterNamespace, filterCluster, remoteEnabled)
  1346. klog.V(4).Infof("ComputeCostDataRange with Key: %s", key)
  1347. // If there is already a request out that uses the same data, wait for it to return to share the results.
  1348. // Otherwise, start executing.
  1349. result, err, _ := cm.RequestGroup.Do(key, func() (interface{}, error) {
  1350. return cm.costDataRange(cli, cp, window, resolution, filterNamespace, filterCluster, remoteEnabled)
  1351. })
  1352. data, ok := result.(map[string]*CostData)
  1353. if !ok {
  1354. return nil, fmt.Errorf("Failed to cast result as map[string]*CostData")
  1355. }
  1356. return data, err
  1357. }
  1358. func (cm *CostModel) costDataRange(cli prometheusClient.Client, cp costAnalyzerCloud.Provider, window kubecost.Window, resolution time.Duration, filterNamespace string, filterCluster string, remoteEnabled bool) (map[string]*CostData, error) {
  1359. clusterID := env.GetClusterID()
  1360. // durHrs := end.Sub(start).Hours() + 1
  1361. if window.IsOpen() {
  1362. return nil, fmt.Errorf("illegal window: %s", window)
  1363. }
  1364. start := *window.Start()
  1365. end := *window.End()
  1366. // Snap resolution to the nearest minute
  1367. resMins := int64(math.Trunc(resolution.Minutes()))
  1368. if resMins == 0 {
  1369. return nil, fmt.Errorf("resolution must be greater than 0.0")
  1370. }
  1371. resolution = time.Duration(resMins) * time.Minute
  1372. // Warn if resolution does not evenly divide window
  1373. if int64(window.Minutes())%int64(resolution.Minutes()) != 0 {
  1374. log.Warningf("CostDataRange: window should be divisible by resolution or else samples may be missed: %s %% %s = %dm", window, resolution, int64(window.Minutes())%int64(resolution.Minutes()))
  1375. }
  1376. // Convert to Prometheus-style duration string in terms of m or h
  1377. resStr := fmt.Sprintf("%dm", resMins)
  1378. if resMins%60 == 0 {
  1379. resStr = fmt.Sprintf("%dh", resMins/60)
  1380. }
  1381. if remoteEnabled {
  1382. remoteLayout := "2006-01-02T15:04:05Z"
  1383. remoteStartStr := window.Start().Format(remoteLayout)
  1384. remoteEndStr := window.End().Format(remoteLayout)
  1385. klog.V(1).Infof("Using remote database for query from %s to %s with window %s", remoteStartStr, remoteEndStr, resolution)
  1386. return CostDataRangeFromSQL("", "", resolution.String(), remoteStartStr, remoteEndStr)
  1387. }
  1388. scrapeIntervalSeconds := cm.ScrapeInterval.Seconds()
  1389. ctx := prom.NewContext(cli)
  1390. queryRAMAlloc := fmt.Sprintf(queryRAMAllocationByteHours, resStr, scrapeIntervalSeconds)
  1391. queryCPUAlloc := fmt.Sprintf(queryCPUAllocationVCPUHours, resStr, scrapeIntervalSeconds)
  1392. queryRAMRequests := fmt.Sprintf(queryRAMRequestsStr, resStr, "", resStr, "")
  1393. queryRAMUsage := fmt.Sprintf(queryRAMUsageStr, resStr, "", resStr, "")
  1394. queryCPURequests := fmt.Sprintf(queryCPURequestsStr, resStr, "", resStr, "")
  1395. queryCPUUsage := fmt.Sprintf(queryCPUUsageStr, resStr, "")
  1396. queryGPURequests := fmt.Sprintf(queryGPURequestsStr, resStr, "", resStr, "", resolution.Hours(), resStr, "")
  1397. queryPVRequests := fmt.Sprintf(queryPVRequestsStr)
  1398. queryPVCAllocation := fmt.Sprintf(queryPVCAllocationFmt, resStr, scrapeIntervalSeconds)
  1399. queryPVHourlyCost := fmt.Sprintf(queryPVHourlyCostFmt, resStr)
  1400. queryNetZoneRequests := fmt.Sprintf(queryZoneNetworkUsage, resStr, "")
  1401. queryNetRegionRequests := fmt.Sprintf(queryRegionNetworkUsage, resStr, "")
  1402. queryNetInternetRequests := fmt.Sprintf(queryInternetNetworkUsage, resStr, "")
  1403. queryNormalization := fmt.Sprintf(normalizationStr, resStr, "")
  1404. // Submit all queries for concurrent evaluation
  1405. resChRAMRequests := ctx.QueryRange(queryRAMRequests, start, end, resolution)
  1406. resChRAMUsage := ctx.QueryRange(queryRAMUsage, start, end, resolution)
  1407. resChRAMAlloc := ctx.QueryRange(queryRAMAlloc, start, end, resolution)
  1408. resChCPURequests := ctx.QueryRange(queryCPURequests, start, end, resolution)
  1409. resChCPUUsage := ctx.QueryRange(queryCPUUsage, start, end, resolution)
  1410. resChCPUAlloc := ctx.QueryRange(queryCPUAlloc, start, end, resolution)
  1411. resChGPURequests := ctx.QueryRange(queryGPURequests, start, end, resolution)
  1412. resChPVRequests := ctx.QueryRange(queryPVRequests, start, end, resolution)
  1413. resChPVCAlloc := ctx.QueryRange(queryPVCAllocation, start, end, resolution)
  1414. resChPVHourlyCost := ctx.QueryRange(queryPVHourlyCost, start, end, resolution)
  1415. resChNetZoneRequests := ctx.QueryRange(queryNetZoneRequests, start, end, resolution)
  1416. resChNetRegionRequests := ctx.QueryRange(queryNetRegionRequests, start, end, resolution)
  1417. resChNetInternetRequests := ctx.QueryRange(queryNetInternetRequests, start, end, resolution)
  1418. resChNSLabels := ctx.QueryRange(fmt.Sprintf(queryNSLabels, resStr), start, end, resolution)
  1419. resChPodLabels := ctx.QueryRange(fmt.Sprintf(queryPodLabels, resStr), start, end, resolution)
  1420. resChNSAnnotations := ctx.QueryRange(fmt.Sprintf(queryNSAnnotations, resStr), start, end, resolution)
  1421. resChPodAnnotations := ctx.QueryRange(fmt.Sprintf(queryPodAnnotations, resStr), start, end, resolution)
  1422. resChServiceLabels := ctx.QueryRange(fmt.Sprintf(queryServiceLabels, resStr), start, end, resolution)
  1423. resChDeploymentLabels := ctx.QueryRange(fmt.Sprintf(queryDeploymentLabels, resStr), start, end, resolution)
  1424. resChStatefulsetLabels := ctx.QueryRange(fmt.Sprintf(queryStatefulsetLabels, resStr), start, end, resolution)
  1425. resChJobs := ctx.QueryRange(queryPodJobs, start, end, resolution)
  1426. resChDaemonsets := ctx.QueryRange(queryPodDaemonsets, start, end, resolution)
  1427. resChNormalization := ctx.QueryRange(queryNormalization, start, end, resolution)
  1428. // Pull k8s pod, controller, service, and namespace details
  1429. podlist := cm.Cache.GetAllPods()
  1430. podDeploymentsMapping, err := getPodDeployments(cm.Cache, podlist, clusterID)
  1431. if err != nil {
  1432. return nil, fmt.Errorf("error querying the kubernetes API: %s", err)
  1433. }
  1434. podStatefulsetsMapping, err := getPodStatefulsets(cm.Cache, podlist, clusterID)
  1435. if err != nil {
  1436. return nil, fmt.Errorf("error querying the kubernetes API: %s", err)
  1437. }
  1438. podServicesMapping, err := getPodServices(cm.Cache, podlist, clusterID)
  1439. if err != nil {
  1440. return nil, fmt.Errorf("error querying the kubernetes API: %s", err)
  1441. }
  1442. namespaceLabelsMapping, err := getNamespaceLabels(cm.Cache, clusterID)
  1443. if err != nil {
  1444. return nil, fmt.Errorf("error querying the kubernetes API: %s", err)
  1445. }
  1446. namespaceAnnotationsMapping, err := getNamespaceAnnotations(cm.Cache, clusterID)
  1447. if err != nil {
  1448. return nil, fmt.Errorf("error querying the kubernetes API: %s", err)
  1449. }
  1450. // Process query results. Handle errors afterwards using ctx.Errors.
  1451. resRAMRequests, _ := resChRAMRequests.Await()
  1452. resRAMUsage, _ := resChRAMUsage.Await()
  1453. resRAMAlloc, _ := resChRAMAlloc.Await()
  1454. resCPURequests, _ := resChCPURequests.Await()
  1455. resCPUUsage, _ := resChCPUUsage.Await()
  1456. resCPUAlloc, _ := resChCPUAlloc.Await()
  1457. resGPURequests, _ := resChGPURequests.Await()
  1458. resPVRequests, _ := resChPVRequests.Await()
  1459. resPVCAlloc, _ := resChPVCAlloc.Await()
  1460. resPVHourlyCost, _ := resChPVHourlyCost.Await()
  1461. resNetZoneRequests, _ := resChNetZoneRequests.Await()
  1462. resNetRegionRequests, _ := resChNetRegionRequests.Await()
  1463. resNetInternetRequests, _ := resChNetInternetRequests.Await()
  1464. resNSLabels, _ := resChNSLabels.Await()
  1465. resPodLabels, _ := resChPodLabels.Await()
  1466. resNSAnnotations, _ := resChNSAnnotations.Await()
  1467. resPodAnnotations, _ := resChPodAnnotations.Await()
  1468. resServiceLabels, _ := resChServiceLabels.Await()
  1469. resDeploymentLabels, _ := resChDeploymentLabels.Await()
  1470. resStatefulsetLabels, _ := resChStatefulsetLabels.Await()
  1471. resDaemonsets, _ := resChDaemonsets.Await()
  1472. resJobs, _ := resChJobs.Await()
  1473. resNormalization, _ := resChNormalization.Await()
  1474. // NOTE: The way we currently handle errors and warnings only early returns if there is an error. Warnings
  1475. // NOTE: will not propagate unless coupled with errors.
  1476. if ctx.HasErrors() {
  1477. // To keep the context of where the errors are occurring, we log the errors here and pass them the error
  1478. // back to the caller. The caller should handle the specific case where error is an ErrorCollection
  1479. for _, promErr := range ctx.Errors() {
  1480. if promErr.Error != nil {
  1481. log.Errorf("CostDataRange: Request Error: %s", promErr.Error)
  1482. }
  1483. if promErr.ParseError != nil {
  1484. log.Errorf("CostDataRange: Parsing Error: %s", promErr.ParseError)
  1485. }
  1486. }
  1487. // ErrorCollection is an collection of errors wrapped in a single error implementation
  1488. return nil, ctx.ErrorCollection()
  1489. }
  1490. normalizationValue, err := getNormalizations(resNormalization)
  1491. if err != nil {
  1492. msg := fmt.Sprintf("error computing normalization for start=%s, end=%s, res=%s", start, end, resolution)
  1493. return nil, prom.WrapError(err, msg)
  1494. }
  1495. pvClaimMapping, err := GetPVInfo(resPVRequests, clusterID)
  1496. if err != nil {
  1497. // Just log for compatibility with KSM less than 1.6
  1498. klog.Infof("Unable to get PV Data: %s", err.Error())
  1499. }
  1500. if pvClaimMapping != nil {
  1501. err = addPVData(cm.Cache, pvClaimMapping, cp)
  1502. if err != nil {
  1503. return nil, fmt.Errorf("pvClaimMapping: %s", err)
  1504. }
  1505. }
  1506. pvCostMapping, err := GetPVCostMetrics(resPVHourlyCost, clusterID)
  1507. if err != nil {
  1508. klog.V(1).Infof("Unable to get PV Hourly Cost Data: %s", err.Error())
  1509. }
  1510. unmountedPVs := make(map[string][]*PersistentVolumeClaimData)
  1511. pvAllocationMapping, err := GetPVAllocationMetrics(resPVCAlloc, clusterID)
  1512. if err != nil {
  1513. klog.V(1).Infof("Unable to get PV Allocation Cost Data: %s", err.Error())
  1514. }
  1515. if pvAllocationMapping != nil {
  1516. addMetricPVData(pvAllocationMapping, pvCostMapping, cp)
  1517. for k, v := range pvAllocationMapping {
  1518. unmountedPVs[k] = v
  1519. }
  1520. }
  1521. nsLabels, err := GetNamespaceLabelsMetrics(resNSLabels, clusterID)
  1522. if err != nil {
  1523. klog.V(1).Infof("Unable to get Namespace Labels for Metrics: %s", err.Error())
  1524. }
  1525. if nsLabels != nil {
  1526. mergeStringMap(namespaceLabelsMapping, nsLabels)
  1527. }
  1528. podLabels, err := GetPodLabelsMetrics(resPodLabels, clusterID)
  1529. if err != nil {
  1530. klog.V(1).Infof("Unable to get Pod Labels for Metrics: %s", err.Error())
  1531. }
  1532. nsAnnotations, err := GetNamespaceAnnotationsMetrics(resNSAnnotations, clusterID)
  1533. if err != nil {
  1534. klog.V(1).Infof("Unable to get Namespace Annotations for Metrics: %s", err.Error())
  1535. }
  1536. if nsAnnotations != nil {
  1537. mergeStringMap(namespaceAnnotationsMapping, nsAnnotations)
  1538. }
  1539. podAnnotations, err := GetPodAnnotationsMetrics(resPodAnnotations, clusterID)
  1540. if err != nil {
  1541. klog.V(1).Infof("Unable to get Pod Annotations for Metrics: %s", err.Error())
  1542. }
  1543. serviceLabels, err := GetServiceSelectorLabelsMetrics(resServiceLabels, clusterID)
  1544. if err != nil {
  1545. klog.V(1).Infof("Unable to get Service Selector Labels for Metrics: %s", err.Error())
  1546. }
  1547. deploymentLabels, err := GetDeploymentMatchLabelsMetrics(resDeploymentLabels, clusterID)
  1548. if err != nil {
  1549. klog.V(1).Infof("Unable to get Deployment Match Labels for Metrics: %s", err.Error())
  1550. }
  1551. statefulsetLabels, err := GetStatefulsetMatchLabelsMetrics(resStatefulsetLabels, clusterID)
  1552. if err != nil {
  1553. klog.V(1).Infof("Unable to get Deployment Match Labels for Metrics: %s", err.Error())
  1554. }
  1555. podStatefulsetMetricsMapping, err := getPodDeploymentsWithMetrics(statefulsetLabels, podLabels)
  1556. if err != nil {
  1557. klog.V(1).Infof("Unable to get match Statefulset Labels Metrics to Pods: %s", err.Error())
  1558. }
  1559. appendLabelsList(podStatefulsetsMapping, podStatefulsetMetricsMapping)
  1560. podDeploymentsMetricsMapping, err := getPodDeploymentsWithMetrics(deploymentLabels, podLabels)
  1561. if err != nil {
  1562. klog.V(1).Infof("Unable to get match Deployment Labels Metrics to Pods: %s", err.Error())
  1563. }
  1564. appendLabelsList(podDeploymentsMapping, podDeploymentsMetricsMapping)
  1565. podDaemonsets, err := GetPodDaemonsetsWithMetrics(resDaemonsets, clusterID)
  1566. if err != nil {
  1567. klog.V(1).Infof("Unable to get Pod Daemonsets for Metrics: %s", err.Error())
  1568. }
  1569. podJobs, err := GetPodJobsWithMetrics(resJobs, clusterID)
  1570. if err != nil {
  1571. klog.V(1).Infof("Unable to get Pod Jobs for Metrics: %s", err.Error())
  1572. }
  1573. podServicesMetricsMapping, err := getPodServicesWithMetrics(serviceLabels, podLabels)
  1574. if err != nil {
  1575. klog.V(1).Infof("Unable to get match Service Labels Metrics to Pods: %s", err.Error())
  1576. }
  1577. appendLabelsList(podServicesMapping, podServicesMetricsMapping)
  1578. networkUsageMap, err := GetNetworkUsageData(resNetZoneRequests, resNetRegionRequests, resNetInternetRequests, clusterID)
  1579. if err != nil {
  1580. klog.V(1).Infof("Unable to get Network Cost Data: %s", err.Error())
  1581. networkUsageMap = make(map[string]*NetworkUsageData)
  1582. }
  1583. containerNameCost := make(map[string]*CostData)
  1584. containers := make(map[string]bool)
  1585. otherClusterPVRecorded := make(map[string]bool)
  1586. RAMReqMap, err := GetNormalizedContainerMetricVectors(resRAMRequests, normalizationValue, clusterID)
  1587. if err != nil {
  1588. return nil, prom.WrapError(err, "GetNormalizedContainerMetricVectors(RAMRequests)")
  1589. }
  1590. for key := range RAMReqMap {
  1591. containers[key] = true
  1592. }
  1593. RAMUsedMap, err := GetNormalizedContainerMetricVectors(resRAMUsage, normalizationValue, clusterID)
  1594. if err != nil {
  1595. return nil, prom.WrapError(err, "GetNormalizedContainerMetricVectors(RAMUsage)")
  1596. }
  1597. for key := range RAMUsedMap {
  1598. containers[key] = true
  1599. }
  1600. CPUReqMap, err := GetNormalizedContainerMetricVectors(resCPURequests, normalizationValue, clusterID)
  1601. if err != nil {
  1602. return nil, prom.WrapError(err, "GetNormalizedContainerMetricVectors(CPURequests)")
  1603. }
  1604. for key := range CPUReqMap {
  1605. containers[key] = true
  1606. }
  1607. // No need to normalize here, as this comes from a counter, namely:
  1608. // rate(container_cpu_usage_seconds_total) which properly accounts for normalized rates
  1609. CPUUsedMap, err := GetContainerMetricVectors(resCPUUsage, clusterID)
  1610. if err != nil {
  1611. return nil, prom.WrapError(err, "GetContainerMetricVectors(CPUUsage)")
  1612. }
  1613. for key := range CPUUsedMap {
  1614. containers[key] = true
  1615. }
  1616. RAMAllocMap, err := GetContainerMetricVectors(resRAMAlloc, clusterID)
  1617. if err != nil {
  1618. return nil, prom.WrapError(err, "GetContainerMetricVectors(RAMAllocations)")
  1619. }
  1620. for key := range RAMAllocMap {
  1621. containers[key] = true
  1622. }
  1623. CPUAllocMap, err := GetContainerMetricVectors(resCPUAlloc, clusterID)
  1624. if err != nil {
  1625. return nil, prom.WrapError(err, "GetContainerMetricVectors(CPUAllocations)")
  1626. }
  1627. for key := range CPUAllocMap {
  1628. containers[key] = true
  1629. }
  1630. GPUReqMap, err := GetNormalizedContainerMetricVectors(resGPURequests, normalizationValue, clusterID)
  1631. if err != nil {
  1632. return nil, prom.WrapError(err, "GetContainerMetricVectors(GPURequests)")
  1633. }
  1634. for key := range GPUReqMap {
  1635. containers[key] = true
  1636. }
  1637. // Request metrics can show up after pod eviction and completion.
  1638. // This method synchronizes requests to allocations such that when
  1639. // allocation is 0, so are requests
  1640. applyAllocationToRequests(RAMAllocMap, RAMReqMap)
  1641. applyAllocationToRequests(CPUAllocMap, CPUReqMap)
  1642. missingNodes := make(map[string]*costAnalyzerCloud.Node)
  1643. missingContainers := make(map[string]*CostData)
  1644. for key := range containers {
  1645. if _, ok := containerNameCost[key]; ok {
  1646. continue // because ordering is important for the allocation model (all PV's applied to the first), just dedupe if it's already been added.
  1647. }
  1648. c, _ := NewContainerMetricFromKey(key)
  1649. RAMReqV, ok := RAMReqMap[key]
  1650. if !ok {
  1651. klog.V(4).Info("no RAM requests for " + key)
  1652. RAMReqV = []*util.Vector{}
  1653. }
  1654. RAMUsedV, ok := RAMUsedMap[key]
  1655. if !ok {
  1656. klog.V(4).Info("no RAM usage for " + key)
  1657. RAMUsedV = []*util.Vector{}
  1658. }
  1659. CPUReqV, ok := CPUReqMap[key]
  1660. if !ok {
  1661. klog.V(4).Info("no CPU requests for " + key)
  1662. CPUReqV = []*util.Vector{}
  1663. }
  1664. CPUUsedV, ok := CPUUsedMap[key]
  1665. if !ok {
  1666. klog.V(4).Info("no CPU usage for " + key)
  1667. CPUUsedV = []*util.Vector{}
  1668. }
  1669. RAMAllocsV, ok := RAMAllocMap[key]
  1670. if !ok {
  1671. klog.V(4).Info("no RAM allocation for " + key)
  1672. RAMAllocsV = []*util.Vector{}
  1673. }
  1674. CPUAllocsV, ok := CPUAllocMap[key]
  1675. if !ok {
  1676. klog.V(4).Info("no CPU allocation for " + key)
  1677. CPUAllocsV = []*util.Vector{}
  1678. }
  1679. GPUReqV, ok := GPUReqMap[key]
  1680. if !ok {
  1681. klog.V(4).Info("no GPU requests for " + key)
  1682. GPUReqV = []*util.Vector{}
  1683. }
  1684. var node *costAnalyzerCloud.Node
  1685. if n, ok := missingNodes[c.NodeName]; ok {
  1686. node = n
  1687. } else {
  1688. node = &costAnalyzerCloud.Node{}
  1689. missingNodes[c.NodeName] = node
  1690. }
  1691. nsKey := c.Namespace + "," + c.ClusterID
  1692. podKey := c.Namespace + "," + c.PodName + "," + c.ClusterID
  1693. namespaceLabels, ok := namespaceLabelsMapping[nsKey]
  1694. if !ok {
  1695. klog.V(4).Infof("Missing data for namespace %s", c.Namespace)
  1696. }
  1697. pLabels := podLabels[podKey]
  1698. if pLabels == nil {
  1699. pLabels = make(map[string]string)
  1700. }
  1701. for k, v := range namespaceLabels {
  1702. if _, ok := pLabels[k]; !ok {
  1703. pLabels[k] = v
  1704. }
  1705. }
  1706. namespaceAnnotations, ok := namespaceAnnotationsMapping[nsKey]
  1707. if !ok {
  1708. klog.V(4).Infof("Missing data for namespace %s", c.Namespace)
  1709. }
  1710. pAnnotations := podAnnotations[podKey]
  1711. if pAnnotations == nil {
  1712. pAnnotations = make(map[string]string)
  1713. }
  1714. for k, v := range namespaceAnnotations {
  1715. if _, ok := pAnnotations[k]; !ok {
  1716. pAnnotations[k] = v
  1717. }
  1718. }
  1719. var podDeployments []string
  1720. if _, ok := podDeploymentsMapping[nsKey]; ok {
  1721. if ds, ok := podDeploymentsMapping[nsKey][c.PodName]; ok {
  1722. podDeployments = ds
  1723. } else {
  1724. podDeployments = []string{}
  1725. }
  1726. }
  1727. var podStatefulSets []string
  1728. if _, ok := podStatefulsetsMapping[nsKey]; ok {
  1729. if ss, ok := podStatefulsetsMapping[nsKey][c.PodName]; ok {
  1730. podStatefulSets = ss
  1731. } else {
  1732. podStatefulSets = []string{}
  1733. }
  1734. }
  1735. var podServices []string
  1736. if _, ok := podServicesMapping[nsKey]; ok {
  1737. if svcs, ok := podServicesMapping[nsKey][c.PodName]; ok {
  1738. podServices = svcs
  1739. } else {
  1740. podServices = []string{}
  1741. }
  1742. }
  1743. var podPVs []*PersistentVolumeClaimData
  1744. var podNetCosts []*util.Vector
  1745. // For PVC data, we'll need to find the claim mapping and cost data. Will need to append
  1746. // cost data since that was populated by cluster data previously. We do this with
  1747. // the pod_pvc_allocation metric
  1748. podPVData, ok := pvAllocationMapping[podKey]
  1749. if !ok {
  1750. klog.V(4).Infof("Failed to locate pv allocation mapping for missing pod.")
  1751. }
  1752. // Delete the current pod key from potentially unmounted pvs
  1753. delete(unmountedPVs, podKey)
  1754. // For network costs, we'll use existing map since it should still contain the
  1755. // correct data.
  1756. var podNetworkCosts []*util.Vector
  1757. if usage, ok := networkUsageMap[podKey]; ok {
  1758. netCosts, err := GetNetworkCost(usage, cp)
  1759. if err != nil {
  1760. klog.V(3).Infof("Error pulling network costs: %s", err.Error())
  1761. } else {
  1762. podNetworkCosts = netCosts
  1763. }
  1764. }
  1765. // Check to see if any other data has been recorded for this namespace, pod, clusterId
  1766. // Follow the pattern of only allowing claims data per pod
  1767. if !otherClusterPVRecorded[podKey] {
  1768. otherClusterPVRecorded[podKey] = true
  1769. podPVs = podPVData
  1770. podNetCosts = podNetworkCosts
  1771. }
  1772. pds := []string{}
  1773. if ds, ok := podDaemonsets[podKey]; ok {
  1774. pds = []string{ds}
  1775. }
  1776. jobs := []string{}
  1777. if job, ok := podJobs[podKey]; ok {
  1778. jobs = []string{job}
  1779. }
  1780. costs := &CostData{
  1781. Name: c.ContainerName,
  1782. PodName: c.PodName,
  1783. NodeName: c.NodeName,
  1784. NodeData: node,
  1785. Namespace: c.Namespace,
  1786. Services: podServices,
  1787. Deployments: podDeployments,
  1788. Daemonsets: pds,
  1789. Statefulsets: podStatefulSets,
  1790. Jobs: jobs,
  1791. RAMReq: RAMReqV,
  1792. RAMUsed: RAMUsedV,
  1793. CPUReq: CPUReqV,
  1794. CPUUsed: CPUUsedV,
  1795. RAMAllocation: RAMAllocsV,
  1796. CPUAllocation: CPUAllocsV,
  1797. GPUReq: GPUReqV,
  1798. Annotations: pAnnotations,
  1799. Labels: pLabels,
  1800. NamespaceLabels: namespaceLabels,
  1801. PVCData: podPVs,
  1802. NetworkData: podNetCosts,
  1803. ClusterID: c.ClusterID,
  1804. ClusterName: cm.ClusterMap.NameFor(c.ClusterID),
  1805. }
  1806. if costDataPassesFilters(cm.ClusterMap, costs, filterNamespace, filterCluster) {
  1807. containerNameCost[key] = costs
  1808. missingContainers[key] = costs
  1809. }
  1810. }
  1811. unmounted := findUnmountedPVCostData(cm.ClusterMap, unmountedPVs, namespaceLabelsMapping, namespaceAnnotationsMapping)
  1812. for k, costs := range unmounted {
  1813. klog.V(4).Infof("Unmounted PVs in Namespace/ClusterID: %s/%s", costs.Namespace, costs.ClusterID)
  1814. if costDataPassesFilters(cm.ClusterMap, costs, filterNamespace, filterCluster) {
  1815. containerNameCost[k] = costs
  1816. }
  1817. }
  1818. if window.Minutes() > 0 {
  1819. dur, off := window.DurationOffsetStrings()
  1820. err = findDeletedNodeInfo(cli, missingNodes, dur, off)
  1821. if err != nil {
  1822. klog.V(1).Infof("Error fetching historical node data: %s", err.Error())
  1823. }
  1824. }
  1825. return containerNameCost, nil
  1826. }
  1827. func applyAllocationToRequests(allocationMap map[string][]*util.Vector, requestMap map[string][]*util.Vector) {
  1828. // The result of the normalize operation will be a new []*util.Vector to replace the requests
  1829. normalizeOp := func(r *util.Vector, x *float64, y *float64) bool {
  1830. // Omit data (return false) if both x and y inputs don't exist
  1831. if x == nil || y == nil {
  1832. return false
  1833. }
  1834. // If the allocation value is 0, 0 out request value
  1835. if *x == 0 {
  1836. r.Value = 0
  1837. } else {
  1838. r.Value = *y
  1839. }
  1840. return true
  1841. }
  1842. // Run normalization on all request vectors in the mapping
  1843. for k, requests := range requestMap {
  1844. // Only run normalization where there are valid allocations
  1845. allocations, ok := allocationMap[k]
  1846. if !ok {
  1847. delete(requestMap, k)
  1848. continue
  1849. }
  1850. // Replace request map with normalized
  1851. requestMap[k] = util.ApplyVectorOp(allocations, requests, normalizeOp)
  1852. }
  1853. }
  1854. func addMetricPVData(pvAllocationMap map[string][]*PersistentVolumeClaimData, pvCostMap map[string]*costAnalyzerCloud.PV, cp costAnalyzerCloud.Provider) {
  1855. cfg, err := cp.GetConfig()
  1856. if err != nil {
  1857. klog.V(1).Infof("Failed to get provider config while adding pv metrics data.")
  1858. return
  1859. }
  1860. for _, pvcDataArray := range pvAllocationMap {
  1861. for _, pvcData := range pvcDataArray {
  1862. costKey := fmt.Sprintf("%s,%s", pvcData.VolumeName, pvcData.ClusterID)
  1863. pvCost, ok := pvCostMap[costKey]
  1864. if !ok {
  1865. pvcData.Volume = &costAnalyzerCloud.PV{
  1866. Cost: cfg.Storage,
  1867. }
  1868. continue
  1869. }
  1870. pvcData.Volume = pvCost
  1871. }
  1872. }
  1873. }
  1874. // Add values that don't already exist in origMap from mergeMap into origMap
  1875. func mergeStringMap(origMap map[string]map[string]string, mergeMap map[string]map[string]string) {
  1876. for k, v := range mergeMap {
  1877. if _, ok := origMap[k]; !ok {
  1878. origMap[k] = v
  1879. }
  1880. }
  1881. }
  1882. func appendLabelsList(mainLabels map[string]map[string][]string, labels map[string]map[string][]string) {
  1883. for k, v := range labels {
  1884. mainLabels[k] = v
  1885. }
  1886. }
  1887. func getNamespaceLabels(cache clustercache.ClusterCache, clusterID string) (map[string]map[string]string, error) {
  1888. nsToLabels := make(map[string]map[string]string)
  1889. nss := cache.GetAllNamespaces()
  1890. for _, ns := range nss {
  1891. labels := make(map[string]string)
  1892. for k, v := range ns.Labels {
  1893. labels[prom.SanitizeLabelName(k)] = v
  1894. }
  1895. nsToLabels[ns.Name+","+clusterID] = labels
  1896. }
  1897. return nsToLabels, nil
  1898. }
  1899. func getNamespaceAnnotations(cache clustercache.ClusterCache, clusterID string) (map[string]map[string]string, error) {
  1900. nsToAnnotations := make(map[string]map[string]string)
  1901. nss := cache.GetAllNamespaces()
  1902. for _, ns := range nss {
  1903. annotations := make(map[string]string)
  1904. for k, v := range ns.Annotations {
  1905. annotations[prom.SanitizeLabelName(k)] = v
  1906. }
  1907. nsToAnnotations[ns.Name+","+clusterID] = annotations
  1908. }
  1909. return nsToAnnotations, nil
  1910. }
  1911. func getDaemonsetsOfPod(pod v1.Pod) []string {
  1912. for _, ownerReference := range pod.ObjectMeta.OwnerReferences {
  1913. if ownerReference.Kind == "DaemonSet" {
  1914. return []string{ownerReference.Name}
  1915. }
  1916. }
  1917. return []string{}
  1918. }
  1919. func getJobsOfPod(pod v1.Pod) []string {
  1920. for _, ownerReference := range pod.ObjectMeta.OwnerReferences {
  1921. if ownerReference.Kind == "Job" {
  1922. return []string{ownerReference.Name}
  1923. }
  1924. }
  1925. return []string{}
  1926. }
  1927. func getStatefulSetsOfPod(pod v1.Pod) []string {
  1928. for _, ownerReference := range pod.ObjectMeta.OwnerReferences {
  1929. if ownerReference.Kind == "StatefulSet" {
  1930. return []string{ownerReference.Name}
  1931. }
  1932. }
  1933. return []string{}
  1934. }
  1935. type PersistentVolumeClaimData struct {
  1936. Class string `json:"class"`
  1937. Claim string `json:"claim"`
  1938. Namespace string `json:"namespace"`
  1939. ClusterID string `json:"clusterId"`
  1940. TimesClaimed int `json:"timesClaimed"`
  1941. VolumeName string `json:"volumeName"`
  1942. Volume *costAnalyzerCloud.PV `json:"persistentVolume"`
  1943. Values []*util.Vector `json:"values"`
  1944. }
  1945. func measureTime(start time.Time, threshold time.Duration, name string) {
  1946. elapsed := time.Since(start)
  1947. if elapsed > threshold {
  1948. klog.V(3).Infof("[Profiler] %s: %s", elapsed, name)
  1949. }
  1950. }
  1951. func measureTimeAsync(start time.Time, threshold time.Duration, name string, ch chan string) {
  1952. elapsed := time.Since(start)
  1953. if elapsed > threshold {
  1954. ch <- fmt.Sprintf("%s took %s", name, time.Since(start))
  1955. }
  1956. }