costmodel.go 77 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226
  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. prometheusClient "github.com/prometheus/client_golang/api"
  18. v1 "k8s.io/api/core/v1"
  19. metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
  20. "k8s.io/apimachinery/pkg/labels"
  21. "k8s.io/klog"
  22. "golang.org/x/sync/singleflight"
  23. )
  24. const (
  25. statusAPIError = 422
  26. profileThreshold = 1000 * 1000 * 1000 // 1s (in ns)
  27. apiPrefix = "/api/v1"
  28. epAlertManagers = apiPrefix + "/alertmanagers"
  29. epLabelValues = apiPrefix + "/label/:name/values"
  30. epSeries = apiPrefix + "/series"
  31. epTargets = apiPrefix + "/targets"
  32. epSnapshot = apiPrefix + "/admin/tsdb/snapshot"
  33. epDeleteSeries = apiPrefix + "/admin/tsdb/delete_series"
  34. epCleanTombstones = apiPrefix + "/admin/tsdb/clean_tombstones"
  35. epConfig = apiPrefix + "/status/config"
  36. epFlags = apiPrefix + "/status/flags"
  37. )
  38. // isCron matches a CronJob name and captures the non-timestamp name
  39. var isCron = regexp.MustCompile(`^(.+)-\d{10}$`)
  40. type CostModel struct {
  41. Cache clustercache.ClusterCache
  42. ClusterMap clusters.ClusterMap
  43. ScrapeInterval time.Duration
  44. RequestGroup *singleflight.Group
  45. pricingMetadata *costAnalyzerCloud.PricingMatchMetadata
  46. }
  47. func NewCostModel(cache clustercache.ClusterCache, clusterMap clusters.ClusterMap, scrapeInterval time.Duration) *CostModel {
  48. // request grouping to prevent over-requesting the same data prior to caching
  49. requestGroup := new(singleflight.Group)
  50. return &CostModel{
  51. Cache: cache,
  52. ClusterMap: clusterMap,
  53. RequestGroup: requestGroup,
  54. ScrapeInterval: scrapeInterval,
  55. }
  56. }
  57. type CostData struct {
  58. Name string `json:"name,omitempty"`
  59. PodName string `json:"podName,omitempty"`
  60. NodeName string `json:"nodeName,omitempty"`
  61. NodeData *costAnalyzerCloud.Node `json:"node,omitempty"`
  62. Namespace string `json:"namespace,omitempty"`
  63. Deployments []string `json:"deployments,omitempty"`
  64. Services []string `json:"services,omitempty"`
  65. Daemonsets []string `json:"daemonsets,omitempty"`
  66. Statefulsets []string `json:"statefulsets,omitempty"`
  67. Jobs []string `json:"jobs,omitempty"`
  68. RAMReq []*util.Vector `json:"ramreq,omitempty"`
  69. RAMUsed []*util.Vector `json:"ramused,omitempty"`
  70. RAMAllocation []*util.Vector `json:"ramallocated,omitempty"`
  71. CPUReq []*util.Vector `json:"cpureq,omitempty"`
  72. CPUUsed []*util.Vector `json:"cpuused,omitempty"`
  73. CPUAllocation []*util.Vector `json:"cpuallocated,omitempty"`
  74. GPUReq []*util.Vector `json:"gpureq,omitempty"`
  75. PVCData []*PersistentVolumeClaimData `json:"pvcData,omitempty"`
  76. NetworkData []*util.Vector `json:"network,omitempty"`
  77. Annotations map[string]string `json:"annotations,omitempty"`
  78. Labels map[string]string `json:"labels,omitempty"`
  79. NamespaceLabels map[string]string `json:"namespaceLabels,omitempty"`
  80. ClusterID string `json:"clusterId"`
  81. ClusterName string `json:"clusterName"`
  82. }
  83. func (cd *CostData) String() string {
  84. 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",
  85. cd.Name, cd.PodName, cd.NodeName, cd.Namespace, strings.Join(cd.Deployments, ", "), strings.Join(cd.Services, ", "),
  86. len(cd.CPUReq), len(cd.CPUUsed), len(cd.CPUAllocation),
  87. len(cd.RAMReq), len(cd.RAMUsed), len(cd.RAMAllocation))
  88. }
  89. func (cd *CostData) GetController() (name string, kind string, hasController bool) {
  90. hasController = false
  91. if len(cd.Deployments) > 0 {
  92. name = cd.Deployments[0]
  93. kind = "deployment"
  94. hasController = true
  95. } else if len(cd.Statefulsets) > 0 {
  96. name = cd.Statefulsets[0]
  97. kind = "statefulset"
  98. hasController = true
  99. } else if len(cd.Daemonsets) > 0 {
  100. name = cd.Daemonsets[0]
  101. kind = "daemonset"
  102. hasController = true
  103. } else if len(cd.Jobs) > 0 {
  104. name = cd.Jobs[0]
  105. kind = "job"
  106. hasController = true
  107. match := isCron.FindStringSubmatch(name)
  108. if match != nil {
  109. name = match[1]
  110. }
  111. }
  112. return name, kind, hasController
  113. }
  114. const (
  115. queryRAMRequestsStr = `avg(
  116. label_replace(
  117. label_replace(
  118. avg(
  119. count_over_time(kube_pod_container_resource_requests_memory_bytes{container!="",container!="POD", node!=""}[%s] %s)
  120. *
  121. avg_over_time(kube_pod_container_resource_requests_memory_bytes{container!="",container!="POD", node!=""}[%s] %s)
  122. ) by (namespace,container,pod,node,cluster_id) , "container_name","$1","container","(.+)"
  123. ), "pod_name","$1","pod","(.+)"
  124. )
  125. ) by (namespace,container_name,pod_name,node,cluster_id)`
  126. queryRAMUsageStr = `sort_desc(
  127. avg(
  128. label_replace(count_over_time(container_memory_working_set_bytes{container_name!="",container_name!="POD", instance!=""}[%s] %s), "node", "$1", "instance","(.+)")
  129. *
  130. label_replace(avg_over_time(container_memory_working_set_bytes{container_name!="",container_name!="POD", instance!=""}[%s] %s), "node", "$1", "instance","(.+)")
  131. ) by (namespace,container_name,pod_name,node,cluster_id)
  132. )`
  133. queryCPURequestsStr = `avg(
  134. label_replace(
  135. label_replace(
  136. avg(
  137. count_over_time(kube_pod_container_resource_requests_cpu_cores{container!="",container!="POD", node!=""}[%s] %s)
  138. *
  139. avg_over_time(kube_pod_container_resource_requests_cpu_cores{container!="",container!="POD", node!=""}[%s] %s)
  140. ) by (namespace,container,pod,node,cluster_id) , "container_name","$1","container","(.+)"
  141. ), "pod_name","$1","pod","(.+)"
  142. )
  143. ) by (namespace,container_name,pod_name,node,cluster_id)`
  144. queryCPUUsageStr = `avg(
  145. label_replace(
  146. rate(
  147. container_cpu_usage_seconds_total{container_name!="",container_name!="POD",instance!=""}[%s] %s
  148. ) , "node", "$1", "instance", "(.+)"
  149. )
  150. ) by (namespace,container_name,pod_name,node,cluster_id)`
  151. queryGPURequestsStr = `avg(
  152. label_replace(
  153. label_replace(
  154. avg(
  155. count_over_time(kube_pod_container_resource_requests{resource="nvidia_com_gpu", container!="",container!="POD", node!=""}[%s] %s)
  156. *
  157. avg_over_time(kube_pod_container_resource_requests{resource="nvidia_com_gpu", container!="",container!="POD", node!=""}[%s] %s)
  158. * %f
  159. ) by (namespace,container,pod,node,cluster_id) , "container_name","$1","container","(.+)"
  160. ), "pod_name","$1","pod","(.+)"
  161. )
  162. ) by (namespace,container_name,pod_name,node,cluster_id)
  163. * 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","(.+)")`
  164. queryPVRequestsStr = `avg(avg(kube_persistentvolumeclaim_info{volumename != ""}) by (persistentvolumeclaim, storageclass, namespace, volumename, cluster_id, kubernetes_node)
  165. *
  166. on (persistentvolumeclaim, namespace, cluster_id, kubernetes_node) group_right(storageclass, volumename)
  167. 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)`
  168. // queryRAMAllocationByteHours yields the total byte-hour RAM allocation over the given
  169. // window, aggregated by container.
  170. // [line 3] sum_over_time(each byte) = [byte*scrape] by metric
  171. // [line 4] (scalar(avg(prometheus_target_interval_length_seconds)) = [seconds/scrape] / 60 / 60 = [hours/scrape] by container
  172. // [lines 2,4] sum(") by unique container key and multiply [byte*scrape] * [hours/scrape] for byte*hours
  173. // [lines 1,5] relabeling
  174. queryRAMAllocationByteHours = `
  175. label_replace(label_replace(
  176. sum(
  177. sum_over_time(container_memory_allocation_bytes{container!="",container!="POD", node!=""}[%s])
  178. ) by (namespace,container,pod,node,cluster_id) * %f / 60 / 60
  179. , "container_name","$1","container","(.+)"), "pod_name","$1","pod","(.+)")`
  180. // queryCPUAllocationVCPUHours yields the total VCPU-hour CPU allocation over the given
  181. // window, aggregated by container.
  182. // [line 3] sum_over_time(each VCPU*mins in window) = [VCPU*scrape] by metric
  183. // [line 4] (scalar(avg(prometheus_target_interval_length_seconds)) = [seconds/scrape] / 60 / 60 = [hours/scrape] by container
  184. // [lines 2,4] sum(") by unique container key and multiply [VCPU*scrape] * [hours/scrape] for VCPU*hours
  185. // [lines 1,5] relabeling
  186. queryCPUAllocationVCPUHours = `
  187. label_replace(label_replace(
  188. sum(
  189. sum_over_time(container_cpu_allocation{container!="",container!="POD", node!=""}[%s])
  190. ) by (namespace,container,pod,node,cluster_id) * %f / 60 / 60
  191. , "container_name","$1","container","(.+)"), "pod_name","$1","pod","(.+)")`
  192. // queryPVCAllocationFmt yields the total byte-hour PVC allocation over the given window.
  193. // sum_over_time(each byte) = [byte*scrape] by metric *(scalar(avg(prometheus_target_interval_length_seconds)) = [seconds/scrape] / 60 / 60 = [hours/scrape] by pod
  194. queryPVCAllocationFmt = `sum(sum_over_time(pod_pvc_allocation[%s])) by (cluster_id, namespace, pod, persistentvolume, persistentvolumeclaim) * %f/60/60`
  195. queryPVHourlyCostFmt = `avg_over_time(pv_hourly_cost[%s])`
  196. queryNSLabels = `avg_over_time(kube_namespace_labels[%s])`
  197. queryPodLabels = `avg_over_time(kube_pod_labels[%s])`
  198. queryNSAnnotations = `avg_over_time(kube_namespace_annotations[%s])`
  199. queryPodAnnotations = `avg_over_time(kube_pod_annotations[%s])`
  200. queryDeploymentLabels = `avg_over_time(deployment_match_labels[%s])`
  201. queryStatefulsetLabels = `avg_over_time(statefulSet_match_labels[%s])`
  202. queryPodDaemonsets = `sum(kube_pod_owner{owner_kind="DaemonSet"}) by (namespace,pod,owner_name,cluster_id)`
  203. queryPodJobs = `sum(kube_pod_owner{owner_kind="Job"}) by (namespace,pod,owner_name,cluster_id)`
  204. queryServiceLabels = `avg_over_time(service_selector_labels[%s])`
  205. 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`
  206. 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`
  207. queryInternetNetworkUsage = `sum(increase(kubecost_pod_network_egress_bytes_total{internet="true"}[%s] %s)) by (namespace,pod_name,cluster_id) / 1024 / 1024 / 1024`
  208. normalizationStr = `max(count_over_time(kube_pod_container_resource_requests_memory_bytes{}[%s] %s))`
  209. )
  210. func (cm *CostModel) ComputeCostData(cli prometheusClient.Client, cp costAnalyzerCloud.Provider, window string, offset string, filterNamespace string) (map[string]*CostData, error) {
  211. queryRAMUsage := fmt.Sprintf(queryRAMUsageStr, window, offset, window, offset)
  212. queryCPUUsage := fmt.Sprintf(queryCPUUsageStr, window, offset)
  213. queryGPURequests := fmt.Sprintf(queryGPURequestsStr, window, offset, window, offset, 1.0, window, offset)
  214. queryPVRequests := fmt.Sprintf(queryPVRequestsStr)
  215. queryNetZoneRequests := fmt.Sprintf(queryZoneNetworkUsage, window, "")
  216. queryNetRegionRequests := fmt.Sprintf(queryRegionNetworkUsage, window, "")
  217. queryNetInternetRequests := fmt.Sprintf(queryInternetNetworkUsage, window, "")
  218. queryNormalization := fmt.Sprintf(normalizationStr, window, offset)
  219. // Cluster ID is specific to the source cluster
  220. clusterID := env.GetClusterID()
  221. // Submit all Prometheus queries asynchronously
  222. ctx := prom.NewContext(cli)
  223. resChRAMUsage := ctx.Query(queryRAMUsage)
  224. resChCPUUsage := ctx.Query(queryCPUUsage)
  225. resChGPURequests := ctx.Query(queryGPURequests)
  226. resChPVRequests := ctx.Query(queryPVRequests)
  227. resChNetZoneRequests := ctx.Query(queryNetZoneRequests)
  228. resChNetRegionRequests := ctx.Query(queryNetRegionRequests)
  229. resChNetInternetRequests := ctx.Query(queryNetInternetRequests)
  230. resChNormalization := ctx.Query(queryNormalization)
  231. // Pull pod information from k8s API
  232. podlist := cm.Cache.GetAllPods()
  233. podDeploymentsMapping, err := getPodDeployments(cm.Cache, podlist, clusterID)
  234. if err != nil {
  235. return nil, err
  236. }
  237. podServicesMapping, err := getPodServices(cm.Cache, podlist, clusterID)
  238. if err != nil {
  239. return nil, err
  240. }
  241. namespaceLabelsMapping, err := getNamespaceLabels(cm.Cache, clusterID)
  242. if err != nil {
  243. return nil, err
  244. }
  245. namespaceAnnotationsMapping, err := getNamespaceAnnotations(cm.Cache, clusterID)
  246. if err != nil {
  247. return nil, err
  248. }
  249. // Process Prometheus query results. Handle errors using ctx.Errors.
  250. resRAMUsage, _ := resChRAMUsage.Await()
  251. resCPUUsage, _ := resChCPUUsage.Await()
  252. resGPURequests, _ := resChGPURequests.Await()
  253. resPVRequests, _ := resChPVRequests.Await()
  254. resNetZoneRequests, _ := resChNetZoneRequests.Await()
  255. resNetRegionRequests, _ := resChNetRegionRequests.Await()
  256. resNetInternetRequests, _ := resChNetInternetRequests.Await()
  257. resNormalization, _ := resChNormalization.Await()
  258. // NOTE: The way we currently handle errors and warnings only early returns if there is an error. Warnings
  259. // NOTE: will not propagate unless coupled with errors.
  260. if ctx.HasErrors() {
  261. // To keep the context of where the errors are occurring, we log the errors here and pass them the error
  262. // back to the caller. The caller should handle the specific case where error is an ErrorCollection
  263. for _, promErr := range ctx.Errors() {
  264. if promErr.Error != nil {
  265. log.Errorf("ComputeCostData: Request Error: %s", promErr.Error)
  266. }
  267. if promErr.ParseError != nil {
  268. log.Errorf("ComputeCostData: Parsing Error: %s", promErr.ParseError)
  269. }
  270. }
  271. // ErrorCollection is an collection of errors wrapped in a single error implementation
  272. // We opt to not return an error for the sake of running as a pure exporter.
  273. log.Warningf("ComputeCostData: continuing despite prometheus errors: %s", ctx.ErrorCollection().Error())
  274. }
  275. defer measureTime(time.Now(), profileThreshold, "ComputeCostData: Processing Query Data")
  276. normalizationValue, err := getNormalization(resNormalization)
  277. if err != nil {
  278. // We opt to not return an error for the sake of running as a pure exporter.
  279. log.Warningf("ComputeCostData: continuing despite error parsing normalization values from %s: %s", queryNormalization, err.Error())
  280. }
  281. nodes, err := cm.GetNodeCost(cp)
  282. if err != nil {
  283. log.Warningf("GetNodeCost: no node cost model available: " + err.Error())
  284. return nil, err
  285. }
  286. // Unmounted PVs represent the PVs that are not mounted or tied to a volume on a container
  287. unmountedPVs := make(map[string][]*PersistentVolumeClaimData)
  288. pvClaimMapping, err := GetPVInfo(resPVRequests, clusterID)
  289. if err != nil {
  290. log.Warningf("GetPVInfo: unable to get PV data: %s", err.Error())
  291. }
  292. if pvClaimMapping != nil {
  293. err = addPVData(cm.Cache, pvClaimMapping, cp)
  294. if err != nil {
  295. return nil, err
  296. }
  297. // copy claim mappings into zombies, then remove as they're discovered
  298. for k, v := range pvClaimMapping {
  299. unmountedPVs[k] = []*PersistentVolumeClaimData{v}
  300. }
  301. }
  302. networkUsageMap, err := GetNetworkUsageData(resNetZoneRequests, resNetRegionRequests, resNetInternetRequests, clusterID)
  303. if err != nil {
  304. klog.V(1).Infof("[Warning] Unable to get Network Cost Data: %s", err.Error())
  305. networkUsageMap = make(map[string]*NetworkUsageData)
  306. }
  307. containerNameCost := make(map[string]*CostData)
  308. containers := make(map[string]bool)
  309. RAMUsedMap, err := GetContainerMetricVector(resRAMUsage, true, normalizationValue, clusterID)
  310. if err != nil {
  311. return nil, err
  312. }
  313. for key := range RAMUsedMap {
  314. containers[key] = true
  315. }
  316. GPUReqMap, err := GetContainerMetricVector(resGPURequests, true, normalizationValue, clusterID)
  317. if err != nil {
  318. return nil, err
  319. }
  320. for key := range GPUReqMap {
  321. containers[key] = true
  322. }
  323. CPUUsedMap, err := GetContainerMetricVector(resCPUUsage, false, 0, clusterID) // No need to normalize here, as this comes from a counter
  324. if err != nil {
  325. return nil, err
  326. }
  327. for key := range CPUUsedMap {
  328. containers[key] = true
  329. }
  330. currentContainers := make(map[string]v1.Pod)
  331. for _, pod := range podlist {
  332. if pod.Status.Phase != v1.PodRunning {
  333. continue
  334. }
  335. cs, err := NewContainerMetricsFromPod(pod, clusterID)
  336. if err != nil {
  337. return nil, err
  338. }
  339. for _, c := range cs {
  340. 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.
  341. currentContainers[c.Key()] = *pod
  342. }
  343. }
  344. missingNodes := make(map[string]*costAnalyzerCloud.Node)
  345. missingContainers := make(map[string]*CostData)
  346. for key := range containers {
  347. if _, ok := containerNameCost[key]; ok {
  348. continue // because ordering is important for the allocation model (all PV's applied to the first), just dedupe if it's already been added.
  349. }
  350. // The _else_ case for this statement is the case in which the container has been
  351. // deleted so we have usage information but not request information. In that case,
  352. // we return partial data for CPU and RAM: only usage and not requests.
  353. if pod, ok := currentContainers[key]; ok {
  354. podName := pod.GetObjectMeta().GetName()
  355. ns := pod.GetObjectMeta().GetNamespace()
  356. nsLabels := namespaceLabelsMapping[ns+","+clusterID]
  357. podLabels := pod.GetObjectMeta().GetLabels()
  358. if podLabels == nil {
  359. podLabels = make(map[string]string)
  360. }
  361. for k, v := range nsLabels {
  362. if _, ok := podLabels[k]; !ok {
  363. podLabels[k] = v
  364. }
  365. }
  366. nsAnnotations := namespaceAnnotationsMapping[ns+","+clusterID]
  367. podAnnotations := pod.GetObjectMeta().GetAnnotations()
  368. if podAnnotations == nil {
  369. podAnnotations = make(map[string]string)
  370. }
  371. for k, v := range nsAnnotations {
  372. if _, ok := podAnnotations[k]; !ok {
  373. podAnnotations[k] = v
  374. }
  375. }
  376. nodeName := pod.Spec.NodeName
  377. var nodeData *costAnalyzerCloud.Node
  378. if _, ok := nodes[nodeName]; ok {
  379. nodeData = nodes[nodeName]
  380. }
  381. nsKey := ns + "," + clusterID
  382. var podDeployments []string
  383. if _, ok := podDeploymentsMapping[nsKey]; ok {
  384. if ds, ok := podDeploymentsMapping[nsKey][pod.GetObjectMeta().GetName()]; ok {
  385. podDeployments = ds
  386. } else {
  387. podDeployments = []string{}
  388. }
  389. }
  390. var podPVs []*PersistentVolumeClaimData
  391. podClaims := pod.Spec.Volumes
  392. for _, vol := range podClaims {
  393. if vol.PersistentVolumeClaim != nil {
  394. name := vol.PersistentVolumeClaim.ClaimName
  395. key := ns + "," + name + "," + clusterID
  396. if pvClaim, ok := pvClaimMapping[key]; ok {
  397. pvClaim.TimesClaimed++
  398. podPVs = append(podPVs, pvClaim)
  399. // Remove entry from potential unmounted pvs
  400. delete(unmountedPVs, key)
  401. }
  402. }
  403. }
  404. var podNetCosts []*util.Vector
  405. if usage, ok := networkUsageMap[ns+","+podName+","+clusterID]; ok {
  406. netCosts, err := GetNetworkCost(usage, cp)
  407. if err != nil {
  408. klog.V(4).Infof("Error pulling network costs: %s", err.Error())
  409. } else {
  410. podNetCosts = netCosts
  411. }
  412. }
  413. var podServices []string
  414. if _, ok := podServicesMapping[nsKey]; ok {
  415. if svcs, ok := podServicesMapping[nsKey][pod.GetObjectMeta().GetName()]; ok {
  416. podServices = svcs
  417. } else {
  418. podServices = []string{}
  419. }
  420. }
  421. for i, container := range pod.Spec.Containers {
  422. containerName := container.Name
  423. // recreate the key and look up data for this container
  424. newKey := NewContainerMetricFromValues(ns, podName, containerName, pod.Spec.NodeName, clusterID).Key()
  425. // k8s.io/apimachinery/pkg/api/resource/amount.go and
  426. // k8s.io/apimachinery/pkg/api/resource/quantity.go for
  427. // details on the "amount" API. See
  428. // https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/#resource-types
  429. // for the units of memory and CPU.
  430. ramRequestBytes := container.Resources.Requests.Memory().Value()
  431. // Because RAM (and CPU) information isn't coming from Prometheus, it won't
  432. // have a timestamp associated with it. We need to provide a timestamp,
  433. // otherwise the vector op that gets applied to take the max of usage
  434. // and request won't work properly and will only take into account
  435. // usage.
  436. RAMReqV := []*util.Vector{
  437. {
  438. Value: float64(ramRequestBytes),
  439. Timestamp: float64(time.Now().UTC().Unix()),
  440. },
  441. }
  442. // use millicores so we can convert to cores in a float64 format
  443. cpuRequestMilliCores := container.Resources.Requests.Cpu().MilliValue()
  444. CPUReqV := []*util.Vector{
  445. {
  446. Value: float64(cpuRequestMilliCores) / 1000,
  447. Timestamp: float64(time.Now().UTC().Unix()),
  448. },
  449. }
  450. RAMUsedV, ok := RAMUsedMap[newKey]
  451. if !ok {
  452. klog.V(4).Info("no RAM usage for " + newKey)
  453. RAMUsedV = []*util.Vector{{}}
  454. }
  455. GPUReqV, ok := GPUReqMap[newKey]
  456. if !ok {
  457. klog.V(4).Info("no GPU requests for " + newKey)
  458. GPUReqV = []*util.Vector{{}}
  459. }
  460. CPUUsedV, ok := CPUUsedMap[newKey]
  461. if !ok {
  462. klog.V(4).Info("no CPU usage for " + newKey)
  463. CPUUsedV = []*util.Vector{{}}
  464. }
  465. var pvReq []*PersistentVolumeClaimData
  466. var netReq []*util.Vector
  467. if i == 0 { // avoid duplicating by just assigning all claims to the first container.
  468. pvReq = podPVs
  469. netReq = podNetCosts
  470. }
  471. costs := &CostData{
  472. Name: containerName,
  473. PodName: podName,
  474. NodeName: nodeName,
  475. Namespace: ns,
  476. Deployments: podDeployments,
  477. Services: podServices,
  478. Daemonsets: getDaemonsetsOfPod(pod),
  479. Jobs: getJobsOfPod(pod),
  480. Statefulsets: getStatefulSetsOfPod(pod),
  481. NodeData: nodeData,
  482. RAMReq: RAMReqV,
  483. RAMUsed: RAMUsedV,
  484. CPUReq: CPUReqV,
  485. CPUUsed: CPUUsedV,
  486. GPUReq: GPUReqV,
  487. PVCData: pvReq,
  488. NetworkData: netReq,
  489. Annotations: podAnnotations,
  490. Labels: podLabels,
  491. NamespaceLabels: nsLabels,
  492. ClusterID: clusterID,
  493. ClusterName: cm.ClusterMap.NameFor(clusterID),
  494. }
  495. costs.CPUAllocation = getContainerAllocation(costs.CPUReq, costs.CPUUsed, "CPU")
  496. costs.RAMAllocation = getContainerAllocation(costs.RAMReq, costs.RAMUsed, "RAM")
  497. if filterNamespace == "" {
  498. containerNameCost[newKey] = costs
  499. } else if costs.Namespace == filterNamespace {
  500. containerNameCost[newKey] = costs
  501. }
  502. }
  503. } else {
  504. // The container has been deleted. Not all information is sent to prometheus via ksm, so fill out what we can without k8s api
  505. klog.V(4).Info("The container " + key + " has been deleted. Calculating allocation but resulting object will be missing data.")
  506. c, err := NewContainerMetricFromKey(key)
  507. if err != nil {
  508. return nil, err
  509. }
  510. // CPU and RAM requests are obtained from the Kubernetes API.
  511. // If this case has been reached, the Kubernetes API will not
  512. // have information about the pod because it no longer exists.
  513. //
  514. // The case where this matters is minimal, mainly in environments
  515. // with very short-lived pods that over-request resources.
  516. RAMReqV := []*util.Vector{{}}
  517. CPUReqV := []*util.Vector{{}}
  518. RAMUsedV, ok := RAMUsedMap[key]
  519. if !ok {
  520. klog.V(4).Info("no RAM usage for " + key)
  521. RAMUsedV = []*util.Vector{{}}
  522. }
  523. GPUReqV, ok := GPUReqMap[key]
  524. if !ok {
  525. klog.V(4).Info("no GPU requests for " + key)
  526. GPUReqV = []*util.Vector{{}}
  527. }
  528. CPUUsedV, ok := CPUUsedMap[key]
  529. if !ok {
  530. klog.V(4).Info("no CPU usage for " + key)
  531. CPUUsedV = []*util.Vector{{}}
  532. }
  533. node, ok := nodes[c.NodeName]
  534. if !ok {
  535. klog.V(4).Infof("Node \"%s\" has been deleted from Kubernetes. Query historical data to get it.", c.NodeName)
  536. if n, ok := missingNodes[c.NodeName]; ok {
  537. node = n
  538. } else {
  539. node = &costAnalyzerCloud.Node{}
  540. missingNodes[c.NodeName] = node
  541. }
  542. }
  543. namespacelabels, ok := namespaceLabelsMapping[c.Namespace+","+c.ClusterID]
  544. if !ok {
  545. klog.V(3).Infof("Missing data for namespace %s", c.Namespace)
  546. }
  547. namespaceAnnotations, ok := namespaceAnnotationsMapping[c.Namespace+","+c.ClusterID]
  548. if !ok {
  549. klog.V(3).Infof("Missing data for namespace %s", c.Namespace)
  550. }
  551. costs := &CostData{
  552. Name: c.ContainerName,
  553. PodName: c.PodName,
  554. NodeName: c.NodeName,
  555. NodeData: node,
  556. Namespace: c.Namespace,
  557. RAMReq: RAMReqV,
  558. RAMUsed: RAMUsedV,
  559. CPUReq: CPUReqV,
  560. CPUUsed: CPUUsedV,
  561. GPUReq: GPUReqV,
  562. Annotations: namespaceAnnotations,
  563. NamespaceLabels: namespacelabels,
  564. ClusterID: c.ClusterID,
  565. ClusterName: cm.ClusterMap.NameFor(c.ClusterID),
  566. }
  567. costs.CPUAllocation = getContainerAllocation(costs.CPUReq, costs.CPUUsed, "CPU")
  568. costs.RAMAllocation = getContainerAllocation(costs.RAMReq, costs.RAMUsed, "RAM")
  569. if filterNamespace == "" {
  570. containerNameCost[key] = costs
  571. missingContainers[key] = costs
  572. } else if costs.Namespace == filterNamespace {
  573. containerNameCost[key] = costs
  574. missingContainers[key] = costs
  575. }
  576. }
  577. }
  578. // Use unmounted pvs to create a mapping of "Unmounted-<Namespace>" containers
  579. // to pass along the cost data
  580. unmounted := findUnmountedPVCostData(cm.ClusterMap, unmountedPVs, namespaceLabelsMapping, namespaceAnnotationsMapping)
  581. for k, costs := range unmounted {
  582. klog.V(4).Infof("Unmounted PVs in Namespace/ClusterID: %s/%s", costs.Namespace, costs.ClusterID)
  583. if filterNamespace == "" {
  584. containerNameCost[k] = costs
  585. } else if costs.Namespace == filterNamespace {
  586. containerNameCost[k] = costs
  587. }
  588. }
  589. err = findDeletedNodeInfo(cli, missingNodes, window, "")
  590. if err != nil {
  591. klog.V(1).Infof("Error fetching historical node data: %s", err.Error())
  592. }
  593. err = findDeletedPodInfo(cli, missingContainers, window)
  594. if err != nil {
  595. klog.V(1).Infof("Error fetching historical pod data: %s", err.Error())
  596. }
  597. return containerNameCost, err
  598. }
  599. func findUnmountedPVCostData(clusterMap clusters.ClusterMap, unmountedPVs map[string][]*PersistentVolumeClaimData, namespaceLabelsMapping map[string]map[string]string, namespaceAnnotationsMapping map[string]map[string]string) map[string]*CostData {
  600. costs := make(map[string]*CostData)
  601. if len(unmountedPVs) == 0 {
  602. return costs
  603. }
  604. for k, pv := range unmountedPVs {
  605. keyParts := strings.Split(k, ",")
  606. if len(keyParts) != 3 {
  607. klog.V(1).Infof("Unmounted PV used key with incorrect parts: %s", k)
  608. continue
  609. }
  610. ns, _, clusterID := keyParts[0], keyParts[1], keyParts[2]
  611. namespacelabels, ok := namespaceLabelsMapping[ns+","+clusterID]
  612. if !ok {
  613. klog.V(3).Infof("Missing data for namespace %s", ns)
  614. }
  615. namespaceAnnotations, ok := namespaceAnnotationsMapping[ns+","+clusterID]
  616. if !ok {
  617. klog.V(3).Infof("Missing data for namespace %s", ns)
  618. }
  619. // Should be a unique "Unmounted" cost data type
  620. name := "unmounted-pvs"
  621. metric := NewContainerMetricFromValues(ns, name, name, "", clusterID)
  622. key := metric.Key()
  623. if costData, ok := costs[key]; !ok {
  624. costs[key] = &CostData{
  625. Name: name,
  626. PodName: name,
  627. NodeName: "",
  628. Annotations: namespaceAnnotations,
  629. Namespace: ns,
  630. NamespaceLabels: namespacelabels,
  631. Labels: namespacelabels,
  632. ClusterID: clusterID,
  633. ClusterName: clusterMap.NameFor(clusterID),
  634. PVCData: pv,
  635. }
  636. } else {
  637. costData.PVCData = append(costData.PVCData, pv...)
  638. }
  639. }
  640. return costs
  641. }
  642. func findDeletedPodInfo(cli prometheusClient.Client, missingContainers map[string]*CostData, window string) error {
  643. if len(missingContainers) > 0 {
  644. queryHistoricalPodLabels := fmt.Sprintf(`kube_pod_labels{}[%s]`, window)
  645. podLabelsResult, _, err := prom.NewContext(cli).QuerySync(queryHistoricalPodLabels)
  646. if err != nil {
  647. log.Errorf("failed to parse historical pod labels: %s", err.Error())
  648. }
  649. podLabels := make(map[string]map[string]string)
  650. if podLabelsResult != nil {
  651. podLabels, err = parsePodLabels(podLabelsResult)
  652. if err != nil {
  653. log.Errorf("failed to parse historical pod labels: %s", err.Error())
  654. }
  655. }
  656. for key, costData := range missingContainers {
  657. cm, _ := NewContainerMetricFromKey(key)
  658. labels, ok := podLabels[cm.PodName]
  659. if !ok {
  660. labels = make(map[string]string)
  661. }
  662. for k, v := range costData.NamespaceLabels {
  663. labels[k] = v
  664. }
  665. costData.Labels = labels
  666. }
  667. }
  668. return nil
  669. }
  670. func findDeletedNodeInfo(cli prometheusClient.Client, missingNodes map[string]*costAnalyzerCloud.Node, window, offset string) error {
  671. if len(missingNodes) > 0 {
  672. defer measureTime(time.Now(), profileThreshold, "Finding Deleted Node Info")
  673. offsetStr := ""
  674. if offset != "" {
  675. offsetStr = fmt.Sprintf("offset %s", offset)
  676. }
  677. queryHistoricalCPUCost := fmt.Sprintf(`avg(avg_over_time(node_cpu_hourly_cost[%s] %s)) by (node, instance, cluster_id)`, window, offsetStr)
  678. queryHistoricalRAMCost := fmt.Sprintf(`avg(avg_over_time(node_ram_hourly_cost[%s] %s)) by (node, instance, cluster_id)`, window, offsetStr)
  679. queryHistoricalGPUCost := fmt.Sprintf(`avg(avg_over_time(node_gpu_hourly_cost[%s] %s)) by (node, instance, cluster_id)`, window, offsetStr)
  680. ctx := prom.NewContext(cli)
  681. cpuCostResCh := ctx.Query(queryHistoricalCPUCost)
  682. ramCostResCh := ctx.Query(queryHistoricalRAMCost)
  683. gpuCostResCh := ctx.Query(queryHistoricalGPUCost)
  684. cpuCostRes, _ := cpuCostResCh.Await()
  685. ramCostRes, _ := ramCostResCh.Await()
  686. gpuCostRes, _ := gpuCostResCh.Await()
  687. if ctx.HasErrors() {
  688. return ctx.ErrorCollection()
  689. }
  690. cpuCosts, err := getCost(cpuCostRes)
  691. if err != nil {
  692. return err
  693. }
  694. ramCosts, err := getCost(ramCostRes)
  695. if err != nil {
  696. return err
  697. }
  698. gpuCosts, err := getCost(gpuCostRes)
  699. if err != nil {
  700. return err
  701. }
  702. if len(cpuCosts) == 0 {
  703. klog.V(1).Infof("Kubecost prometheus metrics not currently available. Ingest this server's /metrics endpoint to get that data.")
  704. }
  705. for node, costv := range cpuCosts {
  706. if _, ok := missingNodes[node]; ok {
  707. missingNodes[node].VCPUCost = fmt.Sprintf("%f", costv[0].Value)
  708. } else {
  709. log.DedupedWarningf(5, "Node `%s` in prometheus but not k8s api", node)
  710. }
  711. }
  712. for node, costv := range ramCosts {
  713. if _, ok := missingNodes[node]; ok {
  714. missingNodes[node].RAMCost = fmt.Sprintf("%f", costv[0].Value)
  715. }
  716. }
  717. for node, costv := range gpuCosts {
  718. if _, ok := missingNodes[node]; ok {
  719. missingNodes[node].GPUCost = fmt.Sprintf("%f", costv[0].Value)
  720. }
  721. }
  722. }
  723. return nil
  724. }
  725. func getContainerAllocation(req []*util.Vector, used []*util.Vector, allocationType string) []*util.Vector {
  726. // The result of the normalize operation will be a new []*util.Vector to replace the requests
  727. allocationOp := func(r *util.Vector, x *float64, y *float64) bool {
  728. if x != nil && y != nil {
  729. x1 := *x
  730. if math.IsNaN(x1) {
  731. klog.V(1).Infof("[Warning] NaN value found during %s allocation calculation for requests.", allocationType)
  732. x1 = 0.0
  733. }
  734. y1 := *y
  735. if math.IsNaN(y1) {
  736. klog.V(1).Infof("[Warning] NaN value found during %s allocation calculation for used.", allocationType)
  737. y1 = 0.0
  738. }
  739. r.Value = math.Max(x1, y1)
  740. } else if x != nil {
  741. r.Value = *x
  742. } else if y != nil {
  743. r.Value = *y
  744. }
  745. return true
  746. }
  747. return util.ApplyVectorOp(req, used, allocationOp)
  748. }
  749. func addPVData(cache clustercache.ClusterCache, pvClaimMapping map[string]*PersistentVolumeClaimData, cloud costAnalyzerCloud.Provider) error {
  750. cfg, err := cloud.GetConfig()
  751. if err != nil {
  752. return err
  753. }
  754. // Pull a region from the first node
  755. var defaultRegion string
  756. nodeList := cache.GetAllNodes()
  757. if len(nodeList) > 0 {
  758. defaultRegion, _ = util.GetRegion(nodeList[0].Labels)
  759. }
  760. storageClasses := cache.GetAllStorageClasses()
  761. storageClassMap := make(map[string]map[string]string)
  762. for _, storageClass := range storageClasses {
  763. params := storageClass.Parameters
  764. storageClassMap[storageClass.ObjectMeta.Name] = params
  765. if storageClass.GetAnnotations()["storageclass.kubernetes.io/is-default-class"] == "true" || storageClass.GetAnnotations()["storageclass.beta.kubernetes.io/is-default-class"] == "true" {
  766. storageClassMap["default"] = params
  767. storageClassMap[""] = params
  768. }
  769. }
  770. pvs := cache.GetAllPersistentVolumes()
  771. pvMap := make(map[string]*costAnalyzerCloud.PV)
  772. for _, pv := range pvs {
  773. parameters, ok := storageClassMap[pv.Spec.StorageClassName]
  774. if !ok {
  775. klog.V(4).Infof("Unable to find parameters for storage class \"%s\". Does pv \"%s\" have a storageClassName?", pv.Spec.StorageClassName, pv.Name)
  776. }
  777. var region string
  778. if r, ok := util.GetRegion(pv.Labels); ok {
  779. region = r
  780. } else {
  781. region = defaultRegion
  782. }
  783. cacPv := &costAnalyzerCloud.PV{
  784. Class: pv.Spec.StorageClassName,
  785. Region: region,
  786. Parameters: parameters,
  787. }
  788. err := GetPVCost(cacPv, pv, cloud, region)
  789. if err != nil {
  790. return err
  791. }
  792. pvMap[pv.Name] = cacPv
  793. }
  794. for _, pvc := range pvClaimMapping {
  795. if vol, ok := pvMap[pvc.VolumeName]; ok {
  796. pvc.Volume = vol
  797. } else {
  798. klog.V(4).Infof("PV not found, using default")
  799. pvc.Volume = &costAnalyzerCloud.PV{
  800. Cost: cfg.Storage,
  801. }
  802. }
  803. }
  804. return nil
  805. }
  806. func GetPVCost(pv *costAnalyzerCloud.PV, kpv *v1.PersistentVolume, cp costAnalyzerCloud.Provider, defaultRegion string) error {
  807. cfg, err := cp.GetConfig()
  808. if err != nil {
  809. return err
  810. }
  811. key := cp.GetPVKey(kpv, pv.Parameters, defaultRegion)
  812. pvWithCost, err := cp.PVPricing(key)
  813. if err != nil {
  814. pv.Cost = cfg.Storage
  815. return err
  816. }
  817. if pvWithCost == nil || pvWithCost.Cost == "" {
  818. pv.Cost = cfg.Storage
  819. return nil // set default cost
  820. }
  821. pv.Cost = pvWithCost.Cost
  822. pv.ProviderID = key.ID()
  823. return nil
  824. }
  825. func (cm *CostModel) GetPricingSourceCounts() (*costAnalyzerCloud.PricingMatchMetadata, error) {
  826. if cm.pricingMetadata != nil {
  827. return cm.pricingMetadata, nil
  828. } else {
  829. return nil, fmt.Errorf("Node costs not yet calculated")
  830. }
  831. }
  832. func (cm *CostModel) GetNodeCost(cp costAnalyzerCloud.Provider) (map[string]*costAnalyzerCloud.Node, error) {
  833. cfg, err := cp.GetConfig()
  834. if err != nil {
  835. return nil, err
  836. }
  837. nodeList := cm.Cache.GetAllNodes()
  838. nodes := make(map[string]*costAnalyzerCloud.Node)
  839. pmd := &costAnalyzerCloud.PricingMatchMetadata{
  840. TotalNodes: 0,
  841. PricingTypeCounts: make(map[costAnalyzerCloud.PricingType]int),
  842. }
  843. for _, n := range nodeList {
  844. name := n.GetObjectMeta().GetName()
  845. nodeLabels := n.GetObjectMeta().GetLabels()
  846. nodeLabels["providerID"] = n.Spec.ProviderID
  847. pmd.TotalNodes++
  848. cnode, err := cp.NodePricing(cp.GetKey(nodeLabels, n))
  849. if err != nil {
  850. klog.Infof("Error getting node pricing. Error: %s", err.Error())
  851. if cnode != nil {
  852. nodes[name] = cnode
  853. continue
  854. } else {
  855. cnode = &costAnalyzerCloud.Node{
  856. VCPUCost: cfg.CPU,
  857. RAMCost: cfg.RAM,
  858. }
  859. }
  860. }
  861. if _, ok := pmd.PricingTypeCounts[cnode.PricingType]; ok {
  862. pmd.PricingTypeCounts[cnode.PricingType]++
  863. } else {
  864. pmd.PricingTypeCounts[cnode.PricingType] = 1
  865. }
  866. newCnode := *cnode
  867. if newCnode.InstanceType == "" {
  868. it, _ := util.GetInstanceType(n.Labels)
  869. newCnode.InstanceType = it
  870. }
  871. if newCnode.Region == "" {
  872. region, _ := util.GetRegion(n.Labels)
  873. newCnode.Region = region
  874. }
  875. newCnode.ProviderID = n.Spec.ProviderID
  876. var cpu float64
  877. if newCnode.VCPU == "" {
  878. cpu = float64(n.Status.Capacity.Cpu().Value())
  879. newCnode.VCPU = n.Status.Capacity.Cpu().String()
  880. } else {
  881. cpu, err = strconv.ParseFloat(newCnode.VCPU, 64)
  882. if err != nil {
  883. klog.V(1).Infof("[Warning] parsing VCPU value: \"%s\" as float64", newCnode.VCPU)
  884. }
  885. }
  886. if math.IsNaN(cpu) {
  887. klog.V(1).Infof("[Warning] cpu parsed as NaN. Setting to 0.")
  888. cpu = 0
  889. }
  890. var ram float64
  891. if newCnode.RAM == "" {
  892. newCnode.RAM = n.Status.Capacity.Memory().String()
  893. }
  894. ram = float64(n.Status.Capacity.Memory().Value())
  895. if math.IsNaN(ram) {
  896. klog.V(1).Infof("[Warning] ram parsed as NaN. Setting to 0.")
  897. ram = 0
  898. }
  899. newCnode.RAMBytes = fmt.Sprintf("%f", ram)
  900. // Azure does not seem to provide a GPU count in its pricing API. GKE supports attaching multiple GPUs
  901. // 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
  902. // 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
  903. gpuc := 0.0
  904. q, ok := n.Status.Capacity["nvidia.com/gpu"]
  905. if ok {
  906. gpuCount := q.Value()
  907. if gpuCount != 0 {
  908. newCnode.GPU = fmt.Sprintf("%d", q.Value())
  909. gpuc = float64(gpuCount)
  910. }
  911. } else {
  912. gpuc, err = strconv.ParseFloat(newCnode.GPU, 64)
  913. if err != nil {
  914. gpuc = 0.0
  915. }
  916. }
  917. if math.IsNaN(gpuc) {
  918. klog.V(1).Infof("[Warning] gpu count parsed as NaN. Setting to 0.")
  919. gpuc = 0.0
  920. }
  921. if newCnode.GPU != "" && newCnode.GPUCost == "" {
  922. // We couldn't find a gpu cost, so fix cpu and ram, then accordingly
  923. klog.V(4).Infof("GPU without cost found for %s, calculating...", cp.GetKey(nodeLabels, n).Features())
  924. defaultCPU, err := strconv.ParseFloat(cfg.CPU, 64)
  925. if err != nil {
  926. klog.V(3).Infof("Could not parse default cpu price")
  927. defaultCPU = 0
  928. }
  929. if math.IsNaN(defaultCPU) {
  930. klog.V(1).Infof("[Warning] defaultCPU parsed as NaN. Setting to 0.")
  931. defaultCPU = 0
  932. }
  933. defaultRAM, err := strconv.ParseFloat(cfg.RAM, 64)
  934. if err != nil {
  935. klog.V(3).Infof("Could not parse default ram price")
  936. defaultRAM = 0
  937. }
  938. if math.IsNaN(defaultRAM) {
  939. klog.V(1).Infof("[Warning] defaultRAM parsed as NaN. Setting to 0.")
  940. defaultRAM = 0
  941. }
  942. defaultGPU, err := strconv.ParseFloat(cfg.GPU, 64)
  943. if err != nil {
  944. klog.V(3).Infof("Could not parse default gpu price")
  945. defaultGPU = 0
  946. }
  947. if math.IsNaN(defaultGPU) {
  948. klog.V(1).Infof("[Warning] defaultGPU parsed as NaN. Setting to 0.")
  949. defaultGPU = 0
  950. }
  951. cpuToRAMRatio := defaultCPU / defaultRAM
  952. if math.IsNaN(cpuToRAMRatio) {
  953. klog.V(1).Infof("[Warning] cpuToRAMRatio[defaultCPU: %f / defaultRam: %f] is NaN. Setting to 0.", defaultCPU, defaultRAM)
  954. cpuToRAMRatio = 0
  955. }
  956. gpuToRAMRatio := defaultGPU / defaultRAM
  957. if math.IsNaN(gpuToRAMRatio) {
  958. klog.V(1).Infof("[Warning] gpuToRAMRatio is NaN. Setting to 0.")
  959. gpuToRAMRatio = 0
  960. }
  961. ramGB := ram / 1024 / 1024 / 1024
  962. if math.IsNaN(ramGB) {
  963. klog.V(1).Infof("[Warning] ramGB is NaN. Setting to 0.")
  964. ramGB = 0
  965. }
  966. ramMultiple := gpuc*gpuToRAMRatio + cpu*cpuToRAMRatio + ramGB
  967. if math.IsNaN(ramMultiple) {
  968. klog.V(1).Infof("[Warning] ramMultiple is NaN. Setting to 0.")
  969. ramMultiple = 0
  970. }
  971. var nodePrice float64
  972. if newCnode.Cost != "" {
  973. nodePrice, err = strconv.ParseFloat(newCnode.Cost, 64)
  974. if err != nil {
  975. klog.V(3).Infof("Could not parse total node price")
  976. return nil, err
  977. }
  978. } else {
  979. nodePrice, err = strconv.ParseFloat(newCnode.VCPUCost, 64) // all the price was allocated the the CPU
  980. if err != nil {
  981. klog.V(3).Infof("Could not parse node vcpu price")
  982. return nil, err
  983. }
  984. }
  985. if math.IsNaN(nodePrice) {
  986. klog.V(1).Infof("[Warning] nodePrice parsed as NaN. Setting to 0.")
  987. nodePrice = 0
  988. }
  989. ramPrice := (nodePrice / ramMultiple)
  990. if math.IsNaN(ramPrice) {
  991. klog.V(1).Infof("[Warning] ramPrice[nodePrice: %f / ramMultiple: %f] parsed as NaN. Setting to 0.", nodePrice, ramMultiple)
  992. ramPrice = 0
  993. }
  994. cpuPrice := ramPrice * cpuToRAMRatio
  995. gpuPrice := ramPrice * gpuToRAMRatio
  996. newCnode.VCPUCost = fmt.Sprintf("%f", cpuPrice)
  997. newCnode.RAMCost = fmt.Sprintf("%f", ramPrice)
  998. newCnode.RAMBytes = fmt.Sprintf("%f", ram)
  999. newCnode.GPUCost = fmt.Sprintf("%f", gpuPrice)
  1000. } else if newCnode.RAMCost == "" {
  1001. // We couldn't find a ramcost, so fix cpu and allocate ram accordingly
  1002. klog.V(4).Infof("No RAM cost found for %s, calculating...", cp.GetKey(nodeLabels, n).Features())
  1003. defaultCPU, err := strconv.ParseFloat(cfg.CPU, 64)
  1004. if err != nil {
  1005. klog.V(3).Infof("Could not parse default cpu price")
  1006. defaultCPU = 0
  1007. }
  1008. if math.IsNaN(defaultCPU) {
  1009. klog.V(1).Infof("[Warning] defaultCPU parsed as NaN. Setting to 0.")
  1010. defaultCPU = 0
  1011. }
  1012. defaultRAM, err := strconv.ParseFloat(cfg.RAM, 64)
  1013. if err != nil {
  1014. klog.V(3).Infof("Could not parse default ram price")
  1015. defaultRAM = 0
  1016. }
  1017. if math.IsNaN(defaultRAM) {
  1018. klog.V(1).Infof("[Warning] defaultRAM parsed as NaN. Setting to 0.")
  1019. defaultRAM = 0
  1020. }
  1021. cpuToRAMRatio := defaultCPU / defaultRAM
  1022. if math.IsNaN(cpuToRAMRatio) {
  1023. klog.V(1).Infof("[Warning] cpuToRAMRatio[defaultCPU: %f / defaultRam: %f] is NaN. Setting to 0.", defaultCPU, defaultRAM)
  1024. cpuToRAMRatio = 0
  1025. }
  1026. ramGB := ram / 1024 / 1024 / 1024
  1027. if math.IsNaN(ramGB) {
  1028. klog.V(1).Infof("[Warning] ramGB is NaN. Setting to 0.")
  1029. ramGB = 0
  1030. }
  1031. ramMultiple := cpu*cpuToRAMRatio + ramGB
  1032. if math.IsNaN(ramMultiple) {
  1033. klog.V(1).Infof("[Warning] ramMultiple is NaN. Setting to 0.")
  1034. ramMultiple = 0
  1035. }
  1036. var nodePrice float64
  1037. if newCnode.Cost != "" {
  1038. nodePrice, err = strconv.ParseFloat(newCnode.Cost, 64)
  1039. if err != nil {
  1040. klog.V(3).Infof("Could not parse total node price")
  1041. return nil, err
  1042. }
  1043. } else {
  1044. nodePrice, err = strconv.ParseFloat(newCnode.VCPUCost, 64) // all the price was allocated the the CPU
  1045. if err != nil {
  1046. klog.V(3).Infof("Could not parse node vcpu price")
  1047. return nil, err
  1048. }
  1049. }
  1050. if math.IsNaN(nodePrice) {
  1051. klog.V(1).Infof("[Warning] nodePrice parsed as NaN. Setting to 0.")
  1052. nodePrice = 0
  1053. }
  1054. ramPrice := (nodePrice / ramMultiple)
  1055. if math.IsNaN(ramPrice) {
  1056. klog.V(1).Infof("[Warning] ramPrice[nodePrice: %f / ramMultiple: %f] parsed as NaN. Setting to 0.", nodePrice, ramMultiple)
  1057. ramPrice = 0
  1058. }
  1059. cpuPrice := ramPrice * cpuToRAMRatio
  1060. if defaultRAM != 0 {
  1061. newCnode.VCPUCost = fmt.Sprintf("%f", cpuPrice)
  1062. newCnode.RAMCost = fmt.Sprintf("%f", ramPrice)
  1063. } else { // just assign the full price to CPU
  1064. if cpu != 0 {
  1065. newCnode.VCPUCost = fmt.Sprintf("%f", nodePrice/cpu)
  1066. } else {
  1067. newCnode.VCPUCost = fmt.Sprintf("%f", nodePrice)
  1068. }
  1069. }
  1070. newCnode.RAMBytes = fmt.Sprintf("%f", ram)
  1071. klog.V(4).Infof("Computed \"%s\" RAM Cost := %v", name, newCnode.RAMCost)
  1072. }
  1073. nodes[name] = &newCnode
  1074. }
  1075. cm.pricingMetadata = pmd
  1076. cp.ApplyReservedInstancePricing(nodes)
  1077. return nodes, nil
  1078. }
  1079. // TODO: drop some logs
  1080. func (cm *CostModel) GetLBCost(cp costAnalyzerCloud.Provider) (map[string]*costAnalyzerCloud.LoadBalancer, error) {
  1081. // for fetching prices from cloud provider
  1082. // cfg, err := cp.GetConfig()
  1083. // if err != nil {
  1084. // return nil, err
  1085. // }
  1086. servicesList := cm.Cache.GetAllServices()
  1087. loadBalancerMap := make(map[string]*costAnalyzerCloud.LoadBalancer)
  1088. for _, service := range servicesList {
  1089. namespace := service.GetObjectMeta().GetNamespace()
  1090. name := service.GetObjectMeta().GetName()
  1091. key := namespace + "," + name // + "," + clusterID?
  1092. if service.Spec.Type == "LoadBalancer" {
  1093. loadBalancer, err := cp.LoadBalancerPricing()
  1094. if err != nil {
  1095. return nil, err
  1096. }
  1097. newLoadBalancer := *loadBalancer
  1098. for _, loadBalancerIngress := range service.Status.LoadBalancer.Ingress {
  1099. newLoadBalancer.IngressIPAddresses = append(newLoadBalancer.IngressIPAddresses, loadBalancerIngress.IP)
  1100. }
  1101. loadBalancerMap[key] = &newLoadBalancer
  1102. }
  1103. }
  1104. return loadBalancerMap, nil
  1105. }
  1106. func getPodServices(cache clustercache.ClusterCache, podList []*v1.Pod, clusterID string) (map[string]map[string][]string, error) {
  1107. servicesList := cache.GetAllServices()
  1108. podServicesMapping := make(map[string]map[string][]string)
  1109. for _, service := range servicesList {
  1110. namespace := service.GetObjectMeta().GetNamespace()
  1111. name := service.GetObjectMeta().GetName()
  1112. key := namespace + "," + clusterID
  1113. if _, ok := podServicesMapping[key]; !ok {
  1114. podServicesMapping[key] = make(map[string][]string)
  1115. }
  1116. s := labels.Nothing()
  1117. if service.Spec.Selector != nil && len(service.Spec.Selector) > 0 {
  1118. s = labels.Set(service.Spec.Selector).AsSelectorPreValidated()
  1119. }
  1120. for _, pod := range podList {
  1121. labelSet := labels.Set(pod.GetObjectMeta().GetLabels())
  1122. if s.Matches(labelSet) && pod.GetObjectMeta().GetNamespace() == namespace {
  1123. services, ok := podServicesMapping[key][pod.GetObjectMeta().GetName()]
  1124. if ok {
  1125. podServicesMapping[key][pod.GetObjectMeta().GetName()] = append(services, name)
  1126. } else {
  1127. podServicesMapping[key][pod.GetObjectMeta().GetName()] = []string{name}
  1128. }
  1129. }
  1130. }
  1131. }
  1132. return podServicesMapping, nil
  1133. }
  1134. func getPodStatefulsets(cache clustercache.ClusterCache, podList []*v1.Pod, clusterID string) (map[string]map[string][]string, error) {
  1135. ssList := cache.GetAllStatefulSets()
  1136. podSSMapping := make(map[string]map[string][]string) // namespace: podName: [deploymentNames]
  1137. for _, ss := range ssList {
  1138. namespace := ss.GetObjectMeta().GetNamespace()
  1139. name := ss.GetObjectMeta().GetName()
  1140. key := namespace + "," + clusterID
  1141. if _, ok := podSSMapping[key]; !ok {
  1142. podSSMapping[key] = make(map[string][]string)
  1143. }
  1144. s, err := metav1.LabelSelectorAsSelector(ss.Spec.Selector)
  1145. if err != nil {
  1146. klog.V(2).Infof("Error doing deployment label conversion: " + err.Error())
  1147. }
  1148. for _, pod := range podList {
  1149. labelSet := labels.Set(pod.GetObjectMeta().GetLabels())
  1150. if s.Matches(labelSet) && pod.GetObjectMeta().GetNamespace() == namespace {
  1151. sss, ok := podSSMapping[key][pod.GetObjectMeta().GetName()]
  1152. if ok {
  1153. podSSMapping[key][pod.GetObjectMeta().GetName()] = append(sss, name)
  1154. } else {
  1155. podSSMapping[key][pod.GetObjectMeta().GetName()] = []string{name}
  1156. }
  1157. }
  1158. }
  1159. }
  1160. return podSSMapping, nil
  1161. }
  1162. func getPodDeployments(cache clustercache.ClusterCache, podList []*v1.Pod, clusterID string) (map[string]map[string][]string, error) {
  1163. deploymentsList := cache.GetAllDeployments()
  1164. podDeploymentsMapping := make(map[string]map[string][]string) // namespace: podName: [deploymentNames]
  1165. for _, deployment := range deploymentsList {
  1166. namespace := deployment.GetObjectMeta().GetNamespace()
  1167. name := deployment.GetObjectMeta().GetName()
  1168. key := namespace + "," + clusterID
  1169. if _, ok := podDeploymentsMapping[key]; !ok {
  1170. podDeploymentsMapping[key] = make(map[string][]string)
  1171. }
  1172. s, err := metav1.LabelSelectorAsSelector(deployment.Spec.Selector)
  1173. if err != nil {
  1174. klog.V(2).Infof("Error doing deployment label conversion: " + err.Error())
  1175. }
  1176. for _, pod := range podList {
  1177. labelSet := labels.Set(pod.GetObjectMeta().GetLabels())
  1178. if s.Matches(labelSet) && pod.GetObjectMeta().GetNamespace() == namespace {
  1179. deployments, ok := podDeploymentsMapping[key][pod.GetObjectMeta().GetName()]
  1180. if ok {
  1181. podDeploymentsMapping[key][pod.GetObjectMeta().GetName()] = append(deployments, name)
  1182. } else {
  1183. podDeploymentsMapping[key][pod.GetObjectMeta().GetName()] = []string{name}
  1184. }
  1185. }
  1186. }
  1187. }
  1188. return podDeploymentsMapping, nil
  1189. }
  1190. func getPodDeploymentsWithMetrics(deploymentLabels map[string]map[string]string, podLabels map[string]map[string]string) (map[string]map[string][]string, error) {
  1191. podDeploymentsMapping := make(map[string]map[string][]string)
  1192. for depKey, depLabels := range deploymentLabels {
  1193. kt, err := NewKeyTuple(depKey)
  1194. if err != nil {
  1195. continue
  1196. }
  1197. namespace := kt.Namespace()
  1198. name := kt.Key()
  1199. clusterID := kt.ClusterID()
  1200. key := namespace + "," + clusterID
  1201. if _, ok := podDeploymentsMapping[key]; !ok {
  1202. podDeploymentsMapping[key] = make(map[string][]string)
  1203. }
  1204. s := labels.Set(depLabels).AsSelectorPreValidated()
  1205. for podKey, pLabels := range podLabels {
  1206. pkey, err := NewKeyTuple(podKey)
  1207. if err != nil {
  1208. continue
  1209. }
  1210. podNamespace := pkey.Namespace()
  1211. podName := pkey.Key()
  1212. podClusterID := pkey.ClusterID()
  1213. labelSet := labels.Set(pLabels)
  1214. if s.Matches(labelSet) && podNamespace == namespace && podClusterID == clusterID {
  1215. deployments, ok := podDeploymentsMapping[key][podName]
  1216. if ok {
  1217. podDeploymentsMapping[key][podName] = append(deployments, name)
  1218. } else {
  1219. podDeploymentsMapping[key][podName] = []string{name}
  1220. }
  1221. }
  1222. }
  1223. }
  1224. // Remove any duplicate data created by metric names
  1225. pruneDuplicateData(podDeploymentsMapping)
  1226. return podDeploymentsMapping, nil
  1227. }
  1228. func getPodServicesWithMetrics(serviceLabels map[string]map[string]string, podLabels map[string]map[string]string) (map[string]map[string][]string, error) {
  1229. podServicesMapping := make(map[string]map[string][]string)
  1230. for servKey, servLabels := range serviceLabels {
  1231. kt, err := NewKeyTuple(servKey)
  1232. if err != nil {
  1233. continue
  1234. }
  1235. namespace := kt.Namespace()
  1236. name := kt.Key()
  1237. clusterID := kt.ClusterID()
  1238. key := namespace + "," + clusterID
  1239. if _, ok := podServicesMapping[key]; !ok {
  1240. podServicesMapping[key] = make(map[string][]string)
  1241. }
  1242. s := labels.Nothing()
  1243. if servLabels != nil && len(servLabels) > 0 {
  1244. s = labels.Set(servLabels).AsSelectorPreValidated()
  1245. }
  1246. for podKey, pLabels := range podLabels {
  1247. pkey, err := NewKeyTuple(podKey)
  1248. if err != nil {
  1249. continue
  1250. }
  1251. podNamespace := pkey.Namespace()
  1252. podName := pkey.Key()
  1253. podClusterID := pkey.ClusterID()
  1254. labelSet := labels.Set(pLabels)
  1255. if s.Matches(labelSet) && podNamespace == namespace && podClusterID == clusterID {
  1256. services, ok := podServicesMapping[key][podName]
  1257. if ok {
  1258. podServicesMapping[key][podName] = append(services, name)
  1259. } else {
  1260. podServicesMapping[key][podName] = []string{name}
  1261. }
  1262. }
  1263. }
  1264. }
  1265. // Remove any duplicate data created by metric names
  1266. pruneDuplicateData(podServicesMapping)
  1267. return podServicesMapping, nil
  1268. }
  1269. // This method alleviates an issue with metrics that used a '_' to replace '-' in deployment
  1270. // and service names. To avoid counting these as multiple deployments/services, we'll remove
  1271. // the '_' version. Not optimal, but takes care of the issue
  1272. func pruneDuplicateData(data map[string]map[string][]string) {
  1273. for _, podMap := range data {
  1274. for podName, values := range podMap {
  1275. podMap[podName] = pruneDuplicates(values)
  1276. }
  1277. }
  1278. }
  1279. // Determine if there is an underscore in the value of a slice. If so, replace _ with -, and then
  1280. // check to see if the result exists in the slice. If both are true, then we DO NOT include that
  1281. // original value in the new slice.
  1282. func pruneDuplicates(s []string) []string {
  1283. m := sliceToSet(s)
  1284. for _, v := range s {
  1285. if strings.Contains(v, "_") {
  1286. name := strings.Replace(v, "_", "-", -1)
  1287. if !m[name] {
  1288. m[name] = true
  1289. }
  1290. delete(m, v)
  1291. }
  1292. }
  1293. return setToSlice(m)
  1294. }
  1295. // Creates a map[string]bool containing the slice values as keys
  1296. func sliceToSet(s []string) map[string]bool {
  1297. m := make(map[string]bool)
  1298. for _, v := range s {
  1299. m[v] = true
  1300. }
  1301. return m
  1302. }
  1303. func setToSlice(m map[string]bool) []string {
  1304. var result []string
  1305. for k := range m {
  1306. result = append(result, k)
  1307. }
  1308. return result
  1309. }
  1310. func costDataPassesFilters(cm clusters.ClusterMap, costs *CostData, namespace string, cluster string) bool {
  1311. passesNamespace := namespace == "" || costs.Namespace == namespace
  1312. passesCluster := cluster == "" || costs.ClusterID == cluster || costs.ClusterName == cluster
  1313. return passesNamespace && passesCluster
  1314. }
  1315. // Finds the a closest multiple less than value
  1316. func floorMultiple(value int64, multiple int64) int64 {
  1317. return (value / multiple) * multiple
  1318. }
  1319. // Attempt to create a key for the request. Reduce the times to minutes in order to more easily group requests based on
  1320. // real time ranges. If for any reason, the key generation fails, return a uuid to ensure uniqueness.
  1321. func requestKeyFor(window kubecost.Window, resolution time.Duration, filterNamespace string, filterCluster string, remoteEnabled bool) string {
  1322. keyLayout := "2006-01-02T15:04Z"
  1323. // We "snap" start time and duration to their closest 5 min multiple less than itself, by
  1324. // applying a snapped duration to a snapped start time.
  1325. durMins := int64(window.Minutes())
  1326. durMins = floorMultiple(durMins, 5)
  1327. sMins := int64(window.Start().Minute())
  1328. sOffset := sMins - floorMultiple(sMins, 5)
  1329. sTime := window.Start().Add(-time.Duration(sOffset) * time.Minute)
  1330. eTime := window.Start().Add(time.Duration(durMins) * time.Minute)
  1331. startKey := sTime.Format(keyLayout)
  1332. endKey := eTime.Format(keyLayout)
  1333. return fmt.Sprintf("%s,%s,%s,%s,%s,%t", startKey, endKey, resolution.String(), filterNamespace, filterCluster, remoteEnabled)
  1334. }
  1335. // func (cm *CostModel) ComputeCostDataRange(cli prometheusClient.Client, cp costAnalyzerCloud.Provider,
  1336. // startString, endString, windowString string, resolutionHours float64, filterNamespace string,
  1337. // filterCluster string, remoteEnabled bool, offset string) (map[string]*CostData, error)
  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. }