costmodel.go 77 KB

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