costmodel.go 78 KB

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