costmodel.go 76 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446
  1. package costmodel
  2. import (
  3. "context"
  4. "encoding/json"
  5. "fmt"
  6. "math"
  7. "net/http"
  8. "os"
  9. "strconv"
  10. "strings"
  11. "sync"
  12. "time"
  13. costAnalyzerCloud "github.com/kubecost/cost-model/cloud"
  14. "github.com/kubecost/cost-model/clustercache"
  15. prometheusClient "github.com/prometheus/client_golang/api"
  16. v1 "k8s.io/api/core/v1"
  17. metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
  18. "k8s.io/apimachinery/pkg/labels"
  19. "k8s.io/client-go/kubernetes"
  20. "k8s.io/klog"
  21. "github.com/google/uuid"
  22. "golang.org/x/sync/singleflight"
  23. )
  24. const (
  25. statusAPIError = 422
  26. apiPrefix = "/api/v1"
  27. epAlertManagers = apiPrefix + "/alertmanagers"
  28. epQuery = apiPrefix + "/query"
  29. epQueryRange = apiPrefix + "/query_range"
  30. epLabelValues = apiPrefix + "/label/:name/values"
  31. epSeries = apiPrefix + "/series"
  32. epTargets = apiPrefix + "/targets"
  33. epSnapshot = apiPrefix + "/admin/tsdb/snapshot"
  34. epDeleteSeries = apiPrefix + "/admin/tsdb/delete_series"
  35. epCleanTombstones = apiPrefix + "/admin/tsdb/clean_tombstones"
  36. epConfig = apiPrefix + "/status/config"
  37. epFlags = apiPrefix + "/status/flags"
  38. clusterIDKey = "CLUSTER_ID"
  39. remoteEnabled = "REMOTE_WRITE_ENABLED"
  40. thanosEnabled = "THANOS_ENABLED"
  41. thanosQueryUrl = "THANOS_QUERY_URL"
  42. )
  43. type CostModel struct {
  44. Cache clustercache.ClusterCache
  45. RequestGroup *singleflight.Group
  46. }
  47. func NewCostModel(cache clustercache.ClusterCache) *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. RequestGroup: requestGroup,
  53. }
  54. }
  55. type CostData struct {
  56. Name string `json:"name,omitempty"`
  57. PodName string `json:"podName,omitempty"`
  58. NodeName string `json:"nodeName,omitempty"`
  59. NodeData *costAnalyzerCloud.Node `json:"node,omitempty"`
  60. Namespace string `json:"namespace,omitempty"`
  61. Deployments []string `json:"deployments,omitempty"`
  62. Services []string `json:"services,omitempty"`
  63. Daemonsets []string `json:"daemonsets,omitempty"`
  64. Statefulsets []string `json:"statefulsets,omitempty"`
  65. Jobs []string `json:"jobs,omitempty"`
  66. RAMReq []*Vector `json:"ramreq,omitempty"`
  67. RAMUsed []*Vector `json:"ramused,omitempty"`
  68. RAMAllocation []*Vector `json:"ramallocated,omitempty"`
  69. CPUReq []*Vector `json:"cpureq,omitempty"`
  70. CPUUsed []*Vector `json:"cpuused,omitempty"`
  71. CPUAllocation []*Vector `json:"cpuallocated,omitempty"`
  72. GPUReq []*Vector `json:"gpureq,omitempty"`
  73. PVCData []*PersistentVolumeClaimData `json:"pvcData,omitempty"`
  74. NetworkData []*Vector `json:"network,omitempty"`
  75. Labels map[string]string `json:"labels,omitempty"`
  76. NamespaceLabels map[string]string `json:"namespaceLabels,omitempty"`
  77. ClusterID string `json:"clusterId"`
  78. }
  79. func (cd *CostData) String() string {
  80. 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",
  81. cd.Name, cd.PodName, cd.NodeName, cd.Namespace, strings.Join(cd.Deployments, ", "), strings.Join(cd.Services, ", "),
  82. len(cd.CPUReq), len(cd.CPUUsed), len(cd.CPUAllocation),
  83. len(cd.RAMReq), len(cd.RAMUsed), len(cd.RAMAllocation))
  84. }
  85. const (
  86. queryRAMRequestsStr = `avg(
  87. label_replace(
  88. label_replace(
  89. avg(
  90. count_over_time(kube_pod_container_resource_requests_memory_bytes{container!="",container!="POD", node!=""}[%s] %s)
  91. *
  92. avg_over_time(kube_pod_container_resource_requests_memory_bytes{container!="",container!="POD", node!=""}[%s] %s)
  93. ) by (namespace,container,pod,node,cluster_id) , "container_name","$1","container","(.+)"
  94. ), "pod_name","$1","pod","(.+)"
  95. )
  96. ) by (namespace,container_name,pod_name,node,cluster_id)`
  97. queryRAMUsageStr = `sort_desc(
  98. avg(
  99. label_replace(count_over_time(container_memory_working_set_bytes{container_name!="",container_name!="POD", instance!=""}[%s] %s), "node", "$1", "instance","(.+)")
  100. *
  101. label_replace(avg_over_time(container_memory_working_set_bytes{container_name!="",container_name!="POD", instance!=""}[%s] %s), "node", "$1", "instance","(.+)")
  102. ) by (namespace,container_name,pod_name,node,cluster_id)
  103. )`
  104. queryCPURequestsStr = `avg(
  105. label_replace(
  106. label_replace(
  107. avg(
  108. count_over_time(kube_pod_container_resource_requests_cpu_cores{container!="",container!="POD", node!=""}[%s] %s)
  109. *
  110. avg_over_time(kube_pod_container_resource_requests_cpu_cores{container!="",container!="POD", node!=""}[%s] %s)
  111. ) by (namespace,container,pod,node,cluster_id) , "container_name","$1","container","(.+)"
  112. ), "pod_name","$1","pod","(.+)"
  113. )
  114. ) by (namespace,container_name,pod_name,node,cluster_id)`
  115. queryCPUUsageStr = `avg(
  116. label_replace(
  117. rate(
  118. container_cpu_usage_seconds_total{container_name!="",container_name!="POD",instance!=""}[%s] %s
  119. ) , "node", "$1", "instance", "(.+)"
  120. )
  121. ) by (namespace,container_name,pod_name,node,cluster_id)`
  122. queryGPURequestsStr = `avg(
  123. label_replace(
  124. label_replace(
  125. avg(
  126. count_over_time(kube_pod_container_resource_requests{resource="nvidia_com_gpu", container!="",container!="POD", node!=""}[%s] %s)
  127. *
  128. avg_over_time(kube_pod_container_resource_requests{resource="nvidia_com_gpu", container!="",container!="POD", node!=""}[%s] %s)
  129. ) by (namespace,container,pod,node,cluster_id) , "container_name","$1","container","(.+)"
  130. ), "pod_name","$1","pod","(.+)"
  131. )
  132. ) by (namespace,container_name,pod_name,node,cluster_id)
  133. * on (pod_name, namespace, node, cluster_id) group_right(container_name) label_replace(avg_over_time(kube_pod_status_phase{phase="Running"}[%s] %s), "pod_name","$1","pod","(.+)")`
  134. queryPVRequestsStr = `avg(kube_persistentvolumeclaim_info) by (persistentvolumeclaim, storageclass, namespace, volumename, cluster_id)
  135. *
  136. on (persistentvolumeclaim, namespace, cluster_id) group_right(storageclass, volumename)
  137. sum(kube_persistentvolumeclaim_resource_requests_storage_bytes) by (persistentvolumeclaim, namespace, cluster_id)`
  138. queryRAMAllocation = `avg(
  139. label_replace(
  140. label_replace(
  141. avg(
  142. count_over_time(container_memory_allocation_bytes{container!="",container!="POD", node!=""}[%s] %s)
  143. *
  144. avg_over_time(container_memory_allocation_bytes{container!="",container!="POD", node!=""}[%s] %s)
  145. ) by (namespace,container,pod,node,cluster_id) , "container_name","$1","container","(.+)"
  146. ), "pod_name","$1","pod","(.+)"
  147. )
  148. ) by (namespace,container_name,pod_name,node,cluster_id)`
  149. queryCPUAllocation = `avg(
  150. label_replace(
  151. label_replace(
  152. avg(
  153. count_over_time(container_cpu_allocation{container!="",container!="POD", node!=""}[%s] %s)
  154. *
  155. avg_over_time(container_cpu_allocation{container!="",container!="POD", node!=""}[%s] %s)
  156. ) by (namespace,container,pod,node,cluster_id) , "container_name","$1","container","(.+)"
  157. ), "pod_name","$1","pod","(.+)"
  158. )
  159. ) by (namespace,container_name,pod_name,node,cluster_id)`
  160. queryPVCAllocation = `avg_over_time(pod_pvc_allocation[%s])`
  161. queryPVHourlyCost = `avg_over_time(pv_hourly_cost[%s])`
  162. queryNSLabels = `avg_over_time(kube_namespace_labels[%s])`
  163. queryPodLabels = `avg_over_time(kube_pod_labels[%s])`
  164. queryDeploymentLabels = `avg_over_time(deployment_match_labels[%s])`
  165. queryStatefulsetLabels = `avg_over_time(statefulSet_match_labels[%s])`
  166. queryServiceLabels = `avg_over_time(service_selector_labels[%s])`
  167. 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`
  168. 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`
  169. queryInternetNetworkUsage = `sum(increase(kubecost_pod_network_egress_bytes_total{internet="true"}[%s] %s)) by (namespace,pod_name,cluster_id) / 1024 / 1024 / 1024`
  170. normalizationStr = `max(count_over_time(kube_pod_container_resource_requests_memory_bytes{}[%s] %s))`
  171. )
  172. type PrometheusMetadata struct {
  173. Running bool `json:"running"`
  174. KubecostDataExists bool `json:"kubecostDataExists"`
  175. }
  176. // ValidatePrometheus tells the model what data prometheus has on it.
  177. func ValidatePrometheus(cli prometheusClient.Client, isThanos bool) (*PrometheusMetadata, error) {
  178. q := "up"
  179. if isThanos {
  180. q += " offset 3h"
  181. }
  182. data, err := Query(cli, q)
  183. if err != nil {
  184. return &PrometheusMetadata{
  185. Running: false,
  186. KubecostDataExists: false,
  187. }, err
  188. }
  189. v, kcmetrics, err := getUptimeData(data)
  190. if err != nil {
  191. return &PrometheusMetadata{
  192. Running: false,
  193. KubecostDataExists: false,
  194. }, err
  195. }
  196. if len(v) > 0 {
  197. return &PrometheusMetadata{
  198. Running: true,
  199. KubecostDataExists: kcmetrics,
  200. }, nil
  201. } else {
  202. return &PrometheusMetadata{
  203. Running: false,
  204. KubecostDataExists: false,
  205. }, fmt.Errorf("No running jobs found on Prometheus at %s", cli.URL(epQuery, nil).Path)
  206. }
  207. }
  208. func getUptimeData(qr interface{}) ([]*Vector, bool, error) {
  209. data, ok := qr.(map[string]interface{})["data"]
  210. if !ok {
  211. e, err := wrapPrometheusError(qr)
  212. if err != nil {
  213. return nil, false, err
  214. }
  215. return nil, false, fmt.Errorf(e)
  216. }
  217. r, ok := data.(map[string]interface{})["result"]
  218. if !ok {
  219. return nil, false, fmt.Errorf("Improperly formatted data from prometheus, data has no result field")
  220. }
  221. results, ok := r.([]interface{})
  222. if !ok {
  223. return nil, false, fmt.Errorf("Improperly formatted results from prometheus, result field is not a slice")
  224. }
  225. jobData := []*Vector{}
  226. kubecostMetrics := false
  227. for _, val := range results {
  228. // For now, just do this for validation. TODO: This can be parsed to figure out the exact running jobs.
  229. metrics, ok := val.(map[string]interface{})["metric"].(map[string]interface{})
  230. if !ok {
  231. return nil, false, fmt.Errorf("Prometheus vector does not have metric labels")
  232. }
  233. jobname, ok := metrics["job"]
  234. if !ok {
  235. return nil, false, fmt.Errorf("up query does not have job names")
  236. }
  237. if jobname == "kubecost" {
  238. kubecostMetrics = true
  239. }
  240. value, ok := val.(map[string]interface{})["value"]
  241. if !ok {
  242. return nil, false, fmt.Errorf("Improperly formatted results from prometheus, value is not a field in the vector")
  243. }
  244. dataPoint, ok := value.([]interface{})
  245. if !ok || len(dataPoint) != 2 {
  246. return nil, false, fmt.Errorf("Improperly formatted datapoint from Prometheus")
  247. }
  248. strVal := dataPoint[1].(string)
  249. v, _ := strconv.ParseFloat(strVal, 64)
  250. toReturn := &Vector{
  251. Timestamp: dataPoint[0].(float64),
  252. Value: v,
  253. }
  254. jobData = append(jobData, toReturn)
  255. }
  256. return jobData, kubecostMetrics, nil
  257. }
  258. func ComputeUptimes(cli prometheusClient.Client) (map[string]float64, error) {
  259. res, err := Query(cli, `container_start_time_seconds{container_name != "POD",container_name != ""}`)
  260. if err != nil {
  261. return nil, err
  262. }
  263. vectors, err := GetContainerMetricVector(res, false, 0, os.Getenv(clusterIDKey))
  264. if err != nil {
  265. return nil, err
  266. }
  267. results := make(map[string]float64)
  268. for key, vector := range vectors {
  269. if err != nil {
  270. return nil, err
  271. }
  272. val := vector[0].Value
  273. uptime := time.Now().Sub(time.Unix(int64(val), 0)).Seconds()
  274. results[key] = uptime
  275. }
  276. return results, nil
  277. }
  278. func (cm *CostModel) ComputeCostData(cli prometheusClient.Client, clientset kubernetes.Interface, cp costAnalyzerCloud.Provider, window string, offset string, filterNamespace string) (map[string]*CostData, error) {
  279. queryRAMRequests := fmt.Sprintf(queryRAMRequestsStr, window, offset, window, offset)
  280. queryRAMUsage := fmt.Sprintf(queryRAMUsageStr, window, offset, window, offset)
  281. queryCPURequests := fmt.Sprintf(queryCPURequestsStr, window, offset, window, offset)
  282. queryCPUUsage := fmt.Sprintf(queryCPUUsageStr, window, offset)
  283. queryGPURequests := fmt.Sprintf(queryGPURequestsStr, window, offset, window, offset, window, offset)
  284. queryPVRequests := fmt.Sprintf(queryPVRequestsStr)
  285. queryNetZoneRequests := fmt.Sprintf(queryZoneNetworkUsage, window, "")
  286. queryNetRegionRequests := fmt.Sprintf(queryRegionNetworkUsage, window, "")
  287. queryNetInternetRequests := fmt.Sprintf(queryInternetNetworkUsage, window, "")
  288. normalization := fmt.Sprintf(normalizationStr, window, offset)
  289. // Cluster ID is specific to the source cluster
  290. clusterID := os.Getenv(clusterIDKey)
  291. var wg sync.WaitGroup
  292. wg.Add(11)
  293. var promErr error
  294. var resultRAMRequests interface{}
  295. go func() {
  296. resultRAMRequests, promErr = Query(cli, queryRAMRequests)
  297. defer wg.Done()
  298. }()
  299. var resultRAMUsage interface{}
  300. go func() {
  301. resultRAMUsage, promErr = Query(cli, queryRAMUsage)
  302. defer wg.Done()
  303. }()
  304. var resultCPURequests interface{}
  305. go func() {
  306. resultCPURequests, promErr = Query(cli, queryCPURequests)
  307. defer wg.Done()
  308. }()
  309. var resultCPUUsage interface{}
  310. go func() {
  311. resultCPUUsage, promErr = Query(cli, queryCPUUsage)
  312. defer wg.Done()
  313. }()
  314. var resultGPURequests interface{}
  315. go func() {
  316. resultGPURequests, promErr = Query(cli, queryGPURequests)
  317. defer wg.Done()
  318. }()
  319. var resultPVRequests interface{}
  320. go func() {
  321. resultPVRequests, promErr = Query(cli, queryPVRequests)
  322. defer wg.Done()
  323. }()
  324. var resultNetZoneRequests interface{}
  325. go func() {
  326. resultNetZoneRequests, promErr = Query(cli, queryNetZoneRequests)
  327. defer wg.Done()
  328. }()
  329. var resultNetRegionRequests interface{}
  330. go func() {
  331. resultNetRegionRequests, promErr = Query(cli, queryNetRegionRequests)
  332. defer wg.Done()
  333. }()
  334. var resultNetInternetRequests interface{}
  335. go func() {
  336. resultNetInternetRequests, promErr = Query(cli, queryNetInternetRequests)
  337. defer wg.Done()
  338. }()
  339. var normalizationResult interface{}
  340. go func() {
  341. normalizationResult, promErr = Query(cli, normalization)
  342. defer wg.Done()
  343. }()
  344. podDeploymentsMapping := make(map[string]map[string][]string)
  345. podServicesMapping := make(map[string]map[string][]string)
  346. namespaceLabelsMapping := make(map[string]map[string]string)
  347. podlist := cm.Cache.GetAllPods()
  348. var k8sErr error
  349. go func() {
  350. defer wg.Done()
  351. podDeploymentsMapping, k8sErr = getPodDeployments(cm.Cache, podlist, clusterID)
  352. if k8sErr != nil {
  353. return
  354. }
  355. podServicesMapping, k8sErr = getPodServices(cm.Cache, podlist, clusterID)
  356. if k8sErr != nil {
  357. return
  358. }
  359. namespaceLabelsMapping, k8sErr = getNamespaceLabels(cm.Cache, clusterID)
  360. if k8sErr != nil {
  361. return
  362. }
  363. }()
  364. wg.Wait()
  365. defer measureTime(time.Now(), "ComputeCostData: Processing Query Data")
  366. if promErr != nil {
  367. return nil, fmt.Errorf("Error querying prometheus: %s", promErr.Error())
  368. }
  369. if k8sErr != nil {
  370. return nil, fmt.Errorf("Error querying the kubernetes api: %s", k8sErr.Error())
  371. }
  372. normalizationValue, err := getNormalization(normalizationResult)
  373. if err != nil {
  374. return nil, fmt.Errorf("Error parsing normalization values: " + err.Error())
  375. }
  376. nodes, err := cm.GetNodeCost(cp)
  377. if err != nil {
  378. klog.V(1).Infof("Warning, no Node cost model available: " + err.Error())
  379. return nil, err
  380. }
  381. pvClaimMapping, err := GetPVInfo(resultPVRequests, clusterID)
  382. if err != nil {
  383. klog.Infof("Unable to get PV Data: %s", err.Error())
  384. }
  385. if pvClaimMapping != nil {
  386. err = addPVData(cm.Cache, pvClaimMapping, cp)
  387. if err != nil {
  388. return nil, err
  389. }
  390. }
  391. networkUsageMap, err := GetNetworkUsageData(resultNetZoneRequests, resultNetRegionRequests, resultNetInternetRequests, clusterID)
  392. if err != nil {
  393. klog.V(1).Infof("Unable to get Network Cost Data: %s", err.Error())
  394. networkUsageMap = make(map[string]*NetworkUsageData)
  395. }
  396. containerNameCost := make(map[string]*CostData)
  397. containers := make(map[string]bool)
  398. RAMReqMap, err := GetContainerMetricVector(resultRAMRequests, true, normalizationValue, clusterID)
  399. if err != nil {
  400. return nil, err
  401. }
  402. for key := range RAMReqMap {
  403. containers[key] = true
  404. }
  405. RAMUsedMap, err := GetContainerMetricVector(resultRAMUsage, true, normalizationValue, clusterID)
  406. if err != nil {
  407. return nil, err
  408. }
  409. for key := range RAMUsedMap {
  410. containers[key] = true
  411. }
  412. CPUReqMap, err := GetContainerMetricVector(resultCPURequests, true, normalizationValue, clusterID)
  413. if err != nil {
  414. return nil, err
  415. }
  416. for key := range CPUReqMap {
  417. containers[key] = true
  418. }
  419. GPUReqMap, err := GetContainerMetricVector(resultGPURequests, true, normalizationValue, clusterID)
  420. if err != nil {
  421. return nil, err
  422. }
  423. for key := range GPUReqMap {
  424. containers[key] = true
  425. }
  426. CPUUsedMap, err := GetContainerMetricVector(resultCPUUsage, false, 0, clusterID) // No need to normalize here, as this comes from a counter
  427. if err != nil {
  428. return nil, err
  429. }
  430. for key := range CPUUsedMap {
  431. containers[key] = true
  432. }
  433. currentContainers := make(map[string]v1.Pod)
  434. for _, pod := range podlist {
  435. if pod.Status.Phase != v1.PodRunning {
  436. continue
  437. }
  438. cs, err := newContainerMetricsFromPod(*pod, clusterID)
  439. if err != nil {
  440. return nil, err
  441. }
  442. for _, c := range cs {
  443. 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.
  444. currentContainers[c.Key()] = *pod
  445. }
  446. }
  447. missingNodes := make(map[string]*costAnalyzerCloud.Node)
  448. missingContainers := make(map[string]*CostData)
  449. for key := range containers {
  450. if _, ok := containerNameCost[key]; ok {
  451. continue // because ordering is important for the allocation model (all PV's applied to the first), just dedupe if it's already been added.
  452. }
  453. if pod, ok := currentContainers[key]; ok {
  454. podName := pod.GetObjectMeta().GetName()
  455. ns := pod.GetObjectMeta().GetNamespace()
  456. nsLabels := namespaceLabelsMapping[ns+","+clusterID]
  457. podLabels := pod.GetObjectMeta().GetLabels()
  458. if podLabels == nil {
  459. podLabels = make(map[string]string)
  460. }
  461. for k, v := range nsLabels {
  462. podLabels[k] = v
  463. }
  464. nodeName := pod.Spec.NodeName
  465. var nodeData *costAnalyzerCloud.Node
  466. if _, ok := nodes[nodeName]; ok {
  467. nodeData = nodes[nodeName]
  468. }
  469. nsKey := ns + "," + clusterID
  470. var podDeployments []string
  471. if _, ok := podDeploymentsMapping[nsKey]; ok {
  472. if ds, ok := podDeploymentsMapping[nsKey][pod.GetObjectMeta().GetName()]; ok {
  473. podDeployments = ds
  474. } else {
  475. podDeployments = []string{}
  476. }
  477. }
  478. var podPVs []*PersistentVolumeClaimData
  479. podClaims := pod.Spec.Volumes
  480. for _, vol := range podClaims {
  481. if vol.PersistentVolumeClaim != nil {
  482. name := vol.PersistentVolumeClaim.ClaimName
  483. if pvClaim, ok := pvClaimMapping[ns+","+name+","+clusterID]; ok {
  484. podPVs = append(podPVs, pvClaim)
  485. }
  486. }
  487. }
  488. var podNetCosts []*Vector
  489. if usage, ok := networkUsageMap[ns+","+podName+","+clusterID]; ok {
  490. netCosts, err := GetNetworkCost(usage, cp)
  491. if err != nil {
  492. klog.V(3).Infof("Error pulling network costs: %s", err.Error())
  493. } else {
  494. podNetCosts = netCosts
  495. }
  496. }
  497. var podServices []string
  498. if _, ok := podServicesMapping[nsKey]; ok {
  499. if svcs, ok := podServicesMapping[nsKey][pod.GetObjectMeta().GetName()]; ok {
  500. podServices = svcs
  501. } else {
  502. podServices = []string{}
  503. }
  504. }
  505. for i, container := range pod.Spec.Containers {
  506. containerName := container.Name
  507. // recreate the key and look up data for this container
  508. newKey := newContainerMetricFromValues(ns, podName, containerName, pod.Spec.NodeName, clusterID).Key()
  509. RAMReqV, ok := RAMReqMap[newKey]
  510. if !ok {
  511. klog.V(4).Info("no RAM requests for " + newKey)
  512. RAMReqV = []*Vector{&Vector{}}
  513. }
  514. RAMUsedV, ok := RAMUsedMap[newKey]
  515. if !ok {
  516. klog.V(4).Info("no RAM usage for " + newKey)
  517. RAMUsedV = []*Vector{&Vector{}}
  518. }
  519. CPUReqV, ok := CPUReqMap[newKey]
  520. if !ok {
  521. klog.V(4).Info("no CPU requests for " + newKey)
  522. CPUReqV = []*Vector{&Vector{}}
  523. }
  524. GPUReqV, ok := GPUReqMap[newKey]
  525. if !ok {
  526. klog.V(4).Info("no GPU requests for " + newKey)
  527. GPUReqV = []*Vector{&Vector{}}
  528. }
  529. CPUUsedV, ok := CPUUsedMap[newKey]
  530. if !ok {
  531. klog.V(4).Info("no CPU usage for " + newKey)
  532. CPUUsedV = []*Vector{&Vector{}}
  533. }
  534. var pvReq []*PersistentVolumeClaimData
  535. var netReq []*Vector
  536. if i == 0 { // avoid duplicating by just assigning all claims to the first container.
  537. pvReq = podPVs
  538. netReq = podNetCosts
  539. }
  540. costs := &CostData{
  541. Name: containerName,
  542. PodName: podName,
  543. NodeName: nodeName,
  544. Namespace: ns,
  545. Deployments: podDeployments,
  546. Services: podServices,
  547. Daemonsets: getDaemonsetsOfPod(pod),
  548. Jobs: getJobsOfPod(pod),
  549. Statefulsets: getStatefulSetsOfPod(pod),
  550. NodeData: nodeData,
  551. RAMReq: RAMReqV,
  552. RAMUsed: RAMUsedV,
  553. CPUReq: CPUReqV,
  554. CPUUsed: CPUUsedV,
  555. GPUReq: GPUReqV,
  556. PVCData: pvReq,
  557. NetworkData: netReq,
  558. Labels: podLabels,
  559. NamespaceLabels: nsLabels,
  560. ClusterID: clusterID,
  561. }
  562. costs.CPUAllocation = getContainerAllocation(costs.CPUReq, costs.CPUUsed)
  563. costs.RAMAllocation = getContainerAllocation(costs.RAMReq, costs.RAMUsed)
  564. if filterNamespace == "" {
  565. containerNameCost[newKey] = costs
  566. } else if costs.Namespace == filterNamespace {
  567. containerNameCost[newKey] = costs
  568. }
  569. }
  570. } else {
  571. // The container has been deleted. Not all information is sent to prometheus via ksm, so fill out what we can without k8s api
  572. klog.V(4).Info("The container " + key + " has been deleted. Calculating allocation but resulting object will be missing data.")
  573. c, err := NewContainerMetricFromKey(key)
  574. if err != nil {
  575. return nil, err
  576. }
  577. RAMReqV, ok := RAMReqMap[key]
  578. if !ok {
  579. klog.V(4).Info("no RAM requests for " + key)
  580. RAMReqV = []*Vector{&Vector{}}
  581. }
  582. RAMUsedV, ok := RAMUsedMap[key]
  583. if !ok {
  584. klog.V(4).Info("no RAM usage for " + key)
  585. RAMUsedV = []*Vector{&Vector{}}
  586. }
  587. CPUReqV, ok := CPUReqMap[key]
  588. if !ok {
  589. klog.V(4).Info("no CPU requests for " + key)
  590. CPUReqV = []*Vector{&Vector{}}
  591. }
  592. GPUReqV, ok := GPUReqMap[key]
  593. if !ok {
  594. klog.V(4).Info("no GPU requests for " + key)
  595. GPUReqV = []*Vector{&Vector{}}
  596. }
  597. CPUUsedV, ok := CPUUsedMap[key]
  598. if !ok {
  599. klog.V(4).Info("no CPU usage for " + key)
  600. CPUUsedV = []*Vector{&Vector{}}
  601. }
  602. node, ok := nodes[c.NodeName]
  603. if !ok {
  604. klog.V(4).Infof("Node \"%s\" has been deleted from Kubernetes. Query historical data to get it.", c.NodeName)
  605. if n, ok := missingNodes[c.NodeName]; ok {
  606. node = n
  607. } else {
  608. node = &costAnalyzerCloud.Node{}
  609. missingNodes[c.NodeName] = node
  610. }
  611. }
  612. namespacelabels, ok := namespaceLabelsMapping[c.Namespace+","+c.ClusterID]
  613. if !ok {
  614. klog.V(3).Infof("Missing data for namespace %s", c.Namespace)
  615. }
  616. costs := &CostData{
  617. Name: c.ContainerName,
  618. PodName: c.PodName,
  619. NodeName: c.NodeName,
  620. NodeData: node,
  621. Namespace: c.Namespace,
  622. RAMReq: RAMReqV,
  623. RAMUsed: RAMUsedV,
  624. CPUReq: CPUReqV,
  625. CPUUsed: CPUUsedV,
  626. GPUReq: GPUReqV,
  627. NamespaceLabels: namespacelabels,
  628. ClusterID: c.ClusterID,
  629. }
  630. costs.CPUAllocation = getContainerAllocation(costs.CPUReq, costs.CPUUsed)
  631. costs.RAMAllocation = getContainerAllocation(costs.RAMReq, costs.RAMUsed)
  632. if filterNamespace == "" {
  633. containerNameCost[key] = costs
  634. missingContainers[key] = costs
  635. } else if costs.Namespace == filterNamespace {
  636. containerNameCost[key] = costs
  637. missingContainers[key] = costs
  638. }
  639. }
  640. }
  641. err = findDeletedNodeInfo(cli, missingNodes, window)
  642. if err != nil {
  643. klog.V(1).Infof("Error fetching historical node data: %s", err.Error())
  644. }
  645. err = findDeletedPodInfo(cli, missingContainers, window)
  646. if err != nil {
  647. klog.V(1).Infof("Error fetching historical pod data: %s", err.Error())
  648. }
  649. return containerNameCost, err
  650. }
  651. func findDeletedPodInfo(cli prometheusClient.Client, missingContainers map[string]*CostData, window string) error {
  652. if len(missingContainers) > 0 {
  653. queryHistoricalPodLabels := fmt.Sprintf(`kube_pod_labels{}[%s]`, window)
  654. podLabelsResult, err := Query(cli, queryHistoricalPodLabels)
  655. if err != nil {
  656. klog.V(1).Infof("Error parsing historical labels: %s", err.Error())
  657. }
  658. podLabels := make(map[string]map[string]string)
  659. if podLabelsResult != nil {
  660. podLabels, err = labelsFromPrometheusQuery(podLabelsResult)
  661. if err != nil {
  662. klog.V(1).Infof("Error parsing historical labels: %s", err.Error())
  663. }
  664. }
  665. for key, costData := range missingContainers {
  666. cm, _ := NewContainerMetricFromKey(key)
  667. labels, ok := podLabels[cm.PodName]
  668. if !ok {
  669. klog.V(1).Infof("Unable to find historical data for pod '%s'", cm.PodName)
  670. labels = make(map[string]string)
  671. }
  672. for k, v := range costData.NamespaceLabels {
  673. labels[k] = v
  674. }
  675. costData.Labels = labels
  676. }
  677. }
  678. return nil
  679. }
  680. func labelsFromPrometheusQuery(qr interface{}) (map[string]map[string]string, error) {
  681. toReturn := make(map[string]map[string]string)
  682. data, ok := qr.(map[string]interface{})["data"]
  683. if !ok {
  684. e, err := wrapPrometheusError(qr)
  685. if err != nil {
  686. return toReturn, err
  687. }
  688. return toReturn, fmt.Errorf(e)
  689. }
  690. for _, val := range data.(map[string]interface{})["result"].([]interface{}) {
  691. metricInterface, ok := val.(map[string]interface{})["metric"]
  692. if !ok {
  693. return toReturn, fmt.Errorf("Metric field does not exist in data result vector")
  694. }
  695. metricMap, ok := metricInterface.(map[string]interface{})
  696. if !ok {
  697. return toReturn, fmt.Errorf("Metric field is improperly formatted")
  698. }
  699. pod, ok := metricMap["pod"]
  700. if !ok {
  701. return toReturn, fmt.Errorf("pod field does not exist in data result vector")
  702. }
  703. podName, ok := pod.(string)
  704. if !ok {
  705. return toReturn, fmt.Errorf("pod field is improperly formatted")
  706. }
  707. for labelName, labelValue := range metricMap {
  708. parsedLabelName := labelName
  709. parsedLv, ok := labelValue.(string)
  710. if !ok {
  711. return toReturn, fmt.Errorf("label value is improperly formatted")
  712. }
  713. if strings.HasPrefix(parsedLabelName, "label_") {
  714. l := strings.Replace(parsedLabelName, "label_", "", 1)
  715. if podLabels, ok := toReturn[podName]; ok {
  716. podLabels[l] = parsedLv
  717. } else {
  718. toReturn[podName] = make(map[string]string)
  719. toReturn[podName][l] = parsedLv
  720. }
  721. }
  722. }
  723. }
  724. return toReturn, nil
  725. }
  726. func findDeletedNodeInfo(cli prometheusClient.Client, missingNodes map[string]*costAnalyzerCloud.Node, window string) error {
  727. if len(missingNodes) > 0 {
  728. defer measureTime(time.Now(), "Finding Deleted Node Info")
  729. q := make([]string, 0, len(missingNodes))
  730. for nodename := range missingNodes {
  731. klog.V(4).Infof("Finding data for deleted node %v", nodename)
  732. q = append(q, nodename)
  733. }
  734. l := strings.Join(q, "|")
  735. queryHistoricalCPUCost := fmt.Sprintf(`avg_over_time(node_cpu_hourly_cost{instance=~"%s"}[%s])`, l, window)
  736. queryHistoricalRAMCost := fmt.Sprintf(`avg_over_time(node_ram_hourly_cost{instance=~"%s"}[%s])`, l, window)
  737. queryHistoricalGPUCost := fmt.Sprintf(`avg_over_time(node_gpu_hourly_cost{instance=~"%s"}[%s])`, l, window)
  738. cpuCostResult, err := Query(cli, queryHistoricalCPUCost)
  739. if err != nil {
  740. return fmt.Errorf("Error fetching cpu cost data: " + err.Error())
  741. }
  742. ramCostResult, err := Query(cli, queryHistoricalRAMCost)
  743. if err != nil {
  744. return fmt.Errorf("Error fetching ram cost data: " + err.Error())
  745. }
  746. gpuCostResult, err := Query(cli, queryHistoricalGPUCost)
  747. if err != nil {
  748. return fmt.Errorf("Error fetching gpu cost data: " + err.Error())
  749. }
  750. cpuCosts, err := getCost(cpuCostResult)
  751. if err != nil {
  752. return err
  753. }
  754. ramCosts, err := getCost(ramCostResult)
  755. if err != nil {
  756. return err
  757. }
  758. gpuCosts, err := getCost(gpuCostResult)
  759. if err != nil {
  760. return err
  761. }
  762. if len(cpuCosts) == 0 {
  763. klog.V(1).Infof("Historical data for node prices not available. Ingest this server's /metrics endpoint to get that data.")
  764. }
  765. for node, costv := range cpuCosts {
  766. if _, ok := missingNodes[node]; ok {
  767. missingNodes[node].VCPUCost = fmt.Sprintf("%f", costv[0].Value)
  768. }
  769. }
  770. for node, costv := range ramCosts {
  771. if _, ok := missingNodes[node]; ok {
  772. missingNodes[node].RAMCost = fmt.Sprintf("%f", costv[0].Value)
  773. }
  774. }
  775. for node, costv := range gpuCosts {
  776. if _, ok := missingNodes[node]; ok {
  777. missingNodes[node].GPUCost = fmt.Sprintf("%f", costv[0].Value)
  778. }
  779. }
  780. }
  781. return nil
  782. }
  783. func getContainerAllocation(req []*Vector, used []*Vector) []*Vector {
  784. // The result of the normalize operation will be a new []*Vector to replace the requests
  785. allocationOp := func(r *Vector, x *float64, y *float64) bool {
  786. if x != nil && y != nil {
  787. r.Value = math.Max(*x, *y)
  788. } else if x != nil {
  789. r.Value = *x
  790. } else if y != nil {
  791. r.Value = *y
  792. }
  793. return true
  794. }
  795. return ApplyVectorOp(req, used, allocationOp)
  796. }
  797. func addPVData(cache clustercache.ClusterCache, pvClaimMapping map[string]*PersistentVolumeClaimData, cloud costAnalyzerCloud.Provider) error {
  798. cfg, err := cloud.GetConfig()
  799. if err != nil {
  800. return err
  801. }
  802. storageClasses := cache.GetAllStorageClasses()
  803. storageClassMap := make(map[string]map[string]string)
  804. for _, storageClass := range storageClasses {
  805. params := storageClass.Parameters
  806. storageClassMap[storageClass.ObjectMeta.Name] = params
  807. if storageClass.GetAnnotations()["storageclass.kubernetes.io/is-default-class"] == "true" || storageClass.GetAnnotations()["storageclass.beta.kubernetes.io/is-default-class"] == "true" {
  808. storageClassMap["default"] = params
  809. storageClassMap[""] = params
  810. }
  811. }
  812. pvs := cache.GetAllPersistentVolumes()
  813. pvMap := make(map[string]*costAnalyzerCloud.PV)
  814. for _, pv := range pvs {
  815. parameters, ok := storageClassMap[pv.Spec.StorageClassName]
  816. if !ok {
  817. klog.V(4).Infof("Unable to find parameters for storage class \"%s\". Does pv \"%s\" have a storageClassName?", pv.Spec.StorageClassName, pv.Name)
  818. }
  819. cacPv := &costAnalyzerCloud.PV{
  820. Class: pv.Spec.StorageClassName,
  821. Region: pv.Labels[v1.LabelZoneRegion],
  822. Parameters: parameters,
  823. }
  824. err := GetPVCost(cacPv, pv, cloud)
  825. if err != nil {
  826. return err
  827. }
  828. pvMap[pv.Name] = cacPv
  829. }
  830. for _, pvc := range pvClaimMapping {
  831. if vol, ok := pvMap[pvc.VolumeName]; ok {
  832. pvc.Volume = vol
  833. } else {
  834. klog.V(4).Infof("PV not found, using default")
  835. pvc.Volume = &costAnalyzerCloud.PV{
  836. Cost: cfg.Storage,
  837. }
  838. }
  839. }
  840. return nil
  841. }
  842. func GetPVCost(pv *costAnalyzerCloud.PV, kpv *v1.PersistentVolume, cp costAnalyzerCloud.Provider) error {
  843. cfg, err := cp.GetConfig()
  844. if err != nil {
  845. return err
  846. }
  847. key := cp.GetPVKey(kpv, pv.Parameters)
  848. pvWithCost, err := cp.PVPricing(key)
  849. if err != nil {
  850. pv.Cost = cfg.Storage
  851. return err
  852. }
  853. if pvWithCost == nil || pvWithCost.Cost == "" {
  854. pv.Cost = cfg.Storage
  855. return nil // set default cost
  856. }
  857. pv.Cost = pvWithCost.Cost
  858. return nil
  859. }
  860. func (cm *CostModel) GetNodeCost(cp costAnalyzerCloud.Provider) (map[string]*costAnalyzerCloud.Node, error) {
  861. cfg, err := cp.GetConfig()
  862. if err != nil {
  863. return nil, err
  864. }
  865. nodeList := cm.Cache.GetAllNodes()
  866. nodes := make(map[string]*costAnalyzerCloud.Node)
  867. for _, n := range nodeList {
  868. name := n.GetObjectMeta().GetName()
  869. nodeLabels := n.GetObjectMeta().GetLabels()
  870. nodeLabels["providerID"] = n.Spec.ProviderID
  871. cnode, err := cp.NodePricing(cp.GetKey(nodeLabels))
  872. if err != nil {
  873. klog.V(1).Infof("Error getting node. Error: " + err.Error())
  874. nodes[name] = cnode
  875. continue
  876. }
  877. newCnode := *cnode
  878. var cpu float64
  879. if newCnode.VCPU == "" {
  880. cpu = float64(n.Status.Capacity.Cpu().Value())
  881. newCnode.VCPU = n.Status.Capacity.Cpu().String()
  882. } else {
  883. cpu, _ = strconv.ParseFloat(newCnode.VCPU, 64)
  884. }
  885. var ram float64
  886. if newCnode.RAM == "" {
  887. newCnode.RAM = n.Status.Capacity.Memory().String()
  888. }
  889. ram = float64(n.Status.Capacity.Memory().Value())
  890. newCnode.RAMBytes = fmt.Sprintf("%f", ram)
  891. if newCnode.GPU != "" && newCnode.GPUCost == "" {
  892. // We couldn't find a gpu cost, so fix cpu and ram, then accordingly
  893. klog.V(4).Infof("GPU without cost found for %s, calculating...", cp.GetKey(nodeLabels).Features())
  894. defaultCPU, err := strconv.ParseFloat(cfg.CPU, 64)
  895. if err != nil {
  896. klog.V(3).Infof("Could not parse default cpu price")
  897. return nil, err
  898. }
  899. defaultRAM, err := strconv.ParseFloat(cfg.RAM, 64)
  900. if err != nil {
  901. klog.V(3).Infof("Could not parse default ram price")
  902. return nil, err
  903. }
  904. defaultGPU, err := strconv.ParseFloat(cfg.GPU, 64)
  905. if err != nil {
  906. klog.V(3).Infof("Could not parse default gpu price")
  907. return nil, err
  908. }
  909. cpuToRAMRatio := defaultCPU / defaultRAM
  910. gpuToRAMRatio := defaultGPU / defaultRAM
  911. ramGB := ram / 1024 / 1024 / 1024
  912. ramMultiple := gpuToRAMRatio + cpu*cpuToRAMRatio + ramGB
  913. var nodePrice float64
  914. if newCnode.Cost != "" {
  915. nodePrice, err = strconv.ParseFloat(newCnode.Cost, 64)
  916. if err != nil {
  917. klog.V(3).Infof("Could not parse total node price")
  918. return nil, err
  919. }
  920. } else {
  921. nodePrice, err = strconv.ParseFloat(newCnode.VCPUCost, 64) // all the price was allocated the the CPU
  922. if err != nil {
  923. klog.V(3).Infof("Could not parse node vcpu price")
  924. return nil, err
  925. }
  926. }
  927. ramPrice := (nodePrice / ramMultiple)
  928. cpuPrice := ramPrice * cpuToRAMRatio
  929. gpuPrice := ramPrice * gpuToRAMRatio
  930. newCnode.VCPUCost = fmt.Sprintf("%f", cpuPrice)
  931. newCnode.RAMCost = fmt.Sprintf("%f", ramPrice)
  932. newCnode.RAMBytes = fmt.Sprintf("%f", ram)
  933. newCnode.GPUCost = fmt.Sprintf("%f", gpuPrice)
  934. } else if newCnode.RAMCost == "" {
  935. // We couldn't find a ramcost, so fix cpu and allocate ram accordingly
  936. klog.V(4).Infof("No RAM cost found for %s, calculating...", cp.GetKey(nodeLabels).Features())
  937. defaultCPU, err := strconv.ParseFloat(cfg.CPU, 64)
  938. if err != nil {
  939. klog.V(3).Infof("Could not parse default cpu price")
  940. return nil, err
  941. }
  942. defaultRAM, err := strconv.ParseFloat(cfg.RAM, 64)
  943. if err != nil {
  944. klog.V(3).Infof("Could not parse default ram price")
  945. return nil, err
  946. }
  947. cpuToRAMRatio := defaultCPU / defaultRAM
  948. ramGB := ram / 1024 / 1024 / 1024
  949. ramMultiple := cpu*cpuToRAMRatio + ramGB
  950. var nodePrice float64
  951. if newCnode.Cost != "" {
  952. nodePrice, err = strconv.ParseFloat(newCnode.Cost, 64)
  953. if err != nil {
  954. klog.V(3).Infof("Could not parse total node price")
  955. return nil, err
  956. }
  957. } else {
  958. nodePrice, err = strconv.ParseFloat(newCnode.VCPUCost, 64) // all the price was allocated the the CPU
  959. if err != nil {
  960. klog.V(3).Infof("Could not parse node vcpu price")
  961. return nil, err
  962. }
  963. }
  964. ramPrice := (nodePrice / ramMultiple)
  965. cpuPrice := ramPrice * cpuToRAMRatio
  966. newCnode.VCPUCost = fmt.Sprintf("%f", cpuPrice)
  967. newCnode.RAMCost = fmt.Sprintf("%f", ramPrice)
  968. newCnode.RAMBytes = fmt.Sprintf("%f", ram)
  969. klog.V(4).Infof("Computed \"%s\" RAM Cost := %v", name, newCnode.RAMCost)
  970. }
  971. nodes[name] = &newCnode
  972. }
  973. cp.ApplyReservedInstancePricing(nodes)
  974. return nodes, nil
  975. }
  976. func getPodServices(cache clustercache.ClusterCache, podList []*v1.Pod, clusterID string) (map[string]map[string][]string, error) {
  977. servicesList := cache.GetAllServices()
  978. podServicesMapping := make(map[string]map[string][]string)
  979. for _, service := range servicesList {
  980. namespace := service.GetObjectMeta().GetNamespace()
  981. name := service.GetObjectMeta().GetName()
  982. key := namespace + "," + clusterID
  983. if _, ok := podServicesMapping[key]; !ok {
  984. podServicesMapping[key] = make(map[string][]string)
  985. }
  986. s := labels.Set(service.Spec.Selector).AsSelectorPreValidated()
  987. for _, pod := range podList {
  988. labelSet := labels.Set(pod.GetObjectMeta().GetLabels())
  989. if s.Matches(labelSet) && pod.GetObjectMeta().GetNamespace() == namespace {
  990. services, ok := podServicesMapping[key][pod.GetObjectMeta().GetName()]
  991. if ok {
  992. podServicesMapping[key][pod.GetObjectMeta().GetName()] = append(services, name)
  993. } else {
  994. podServicesMapping[key][pod.GetObjectMeta().GetName()] = []string{name}
  995. }
  996. }
  997. }
  998. }
  999. return podServicesMapping, nil
  1000. }
  1001. func getPodStatefulsets(cache clustercache.ClusterCache, podList []*v1.Pod, clusterID string) (map[string]map[string][]string, error) {
  1002. ssList := cache.GetAllStatefulSets()
  1003. podSSMapping := make(map[string]map[string][]string) // namespace: podName: [deploymentNames]
  1004. for _, ss := range ssList {
  1005. namespace := ss.GetObjectMeta().GetNamespace()
  1006. name := ss.GetObjectMeta().GetName()
  1007. key := namespace + "," + clusterID
  1008. if _, ok := podSSMapping[key]; !ok {
  1009. podSSMapping[key] = make(map[string][]string)
  1010. }
  1011. s, err := metav1.LabelSelectorAsSelector(ss.Spec.Selector)
  1012. if err != nil {
  1013. klog.V(2).Infof("Error doing deployment label conversion: " + err.Error())
  1014. }
  1015. for _, pod := range podList {
  1016. labelSet := labels.Set(pod.GetObjectMeta().GetLabels())
  1017. if s.Matches(labelSet) && pod.GetObjectMeta().GetNamespace() == namespace {
  1018. sss, ok := podSSMapping[key][pod.GetObjectMeta().GetName()]
  1019. if ok {
  1020. podSSMapping[key][pod.GetObjectMeta().GetName()] = append(sss, name)
  1021. } else {
  1022. podSSMapping[key][pod.GetObjectMeta().GetName()] = []string{name}
  1023. }
  1024. }
  1025. }
  1026. }
  1027. return podSSMapping, nil
  1028. }
  1029. func getPodDeployments(cache clustercache.ClusterCache, podList []*v1.Pod, clusterID string) (map[string]map[string][]string, error) {
  1030. deploymentsList := cache.GetAllDeployments()
  1031. podDeploymentsMapping := make(map[string]map[string][]string) // namespace: podName: [deploymentNames]
  1032. for _, deployment := range deploymentsList {
  1033. namespace := deployment.GetObjectMeta().GetNamespace()
  1034. name := deployment.GetObjectMeta().GetName()
  1035. key := namespace + "," + clusterID
  1036. if _, ok := podDeploymentsMapping[key]; !ok {
  1037. podDeploymentsMapping[key] = make(map[string][]string)
  1038. }
  1039. s, err := metav1.LabelSelectorAsSelector(deployment.Spec.Selector)
  1040. if err != nil {
  1041. klog.V(2).Infof("Error doing deployment label conversion: " + err.Error())
  1042. }
  1043. for _, pod := range podList {
  1044. labelSet := labels.Set(pod.GetObjectMeta().GetLabels())
  1045. if s.Matches(labelSet) && pod.GetObjectMeta().GetNamespace() == namespace {
  1046. deployments, ok := podDeploymentsMapping[key][pod.GetObjectMeta().GetName()]
  1047. if ok {
  1048. podDeploymentsMapping[key][pod.GetObjectMeta().GetName()] = append(deployments, name)
  1049. } else {
  1050. podDeploymentsMapping[key][pod.GetObjectMeta().GetName()] = []string{name}
  1051. }
  1052. }
  1053. }
  1054. }
  1055. return podDeploymentsMapping, nil
  1056. }
  1057. func getPodDeploymentsWithMetrics(deploymentLabels map[string]map[string]string, podLabels map[string]map[string]string) (map[string]map[string][]string, error) {
  1058. podDeploymentsMapping := make(map[string]map[string][]string)
  1059. for depKey, depLabels := range deploymentLabels {
  1060. kt, err := NewKeyTuple(depKey)
  1061. if err != nil {
  1062. continue
  1063. }
  1064. namespace := kt.Namespace
  1065. name := kt.Key
  1066. clusterID := kt.ClusterID
  1067. key := namespace + "," + clusterID
  1068. if _, ok := podDeploymentsMapping[key]; !ok {
  1069. podDeploymentsMapping[key] = make(map[string][]string)
  1070. }
  1071. s := labels.Set(depLabels).AsSelectorPreValidated()
  1072. for podKey, pLabels := range podLabels {
  1073. pkey, err := NewKeyTuple(podKey)
  1074. if err != nil {
  1075. continue
  1076. }
  1077. podNamespace := pkey.Namespace
  1078. podName := pkey.Key
  1079. podClusterID := pkey.ClusterID
  1080. labelSet := labels.Set(pLabels)
  1081. if s.Matches(labelSet) && podNamespace == namespace && podClusterID == clusterID {
  1082. deployments, ok := podDeploymentsMapping[key][podName]
  1083. if ok {
  1084. podDeploymentsMapping[key][podName] = append(deployments, name)
  1085. } else {
  1086. podDeploymentsMapping[key][podName] = []string{name}
  1087. }
  1088. }
  1089. }
  1090. }
  1091. // Remove any duplicate data created by metric names
  1092. pruneDuplicateData(podDeploymentsMapping)
  1093. return podDeploymentsMapping, nil
  1094. }
  1095. func getPodServicesWithMetrics(serviceLabels map[string]map[string]string, podLabels map[string]map[string]string) (map[string]map[string][]string, error) {
  1096. podServicesMapping := make(map[string]map[string][]string)
  1097. for servKey, servLabels := range serviceLabels {
  1098. kt, err := NewKeyTuple(servKey)
  1099. if err != nil {
  1100. continue
  1101. }
  1102. namespace := kt.Namespace
  1103. name := kt.Key
  1104. clusterID := kt.ClusterID
  1105. key := namespace + "," + clusterID
  1106. if _, ok := podServicesMapping[key]; !ok {
  1107. podServicesMapping[key] = make(map[string][]string)
  1108. }
  1109. s := labels.Set(servLabels).AsSelectorPreValidated()
  1110. for podKey, pLabels := range podLabels {
  1111. pkey, err := NewKeyTuple(podKey)
  1112. if err != nil {
  1113. continue
  1114. }
  1115. podNamespace := pkey.Namespace
  1116. podName := pkey.Key
  1117. podClusterID := pkey.ClusterID
  1118. labelSet := labels.Set(pLabels)
  1119. if s.Matches(labelSet) && podNamespace == namespace && podClusterID == clusterID {
  1120. services, ok := podServicesMapping[key][podName]
  1121. if ok {
  1122. podServicesMapping[key][podName] = append(services, name)
  1123. } else {
  1124. podServicesMapping[key][podName] = []string{name}
  1125. }
  1126. }
  1127. }
  1128. }
  1129. // Remove any duplicate data created by metric names
  1130. pruneDuplicateData(podServicesMapping)
  1131. return podServicesMapping, nil
  1132. }
  1133. // This method alleviates an issue with metrics that used a '_' to replace '-' in deployment
  1134. // and service names. To avoid counting these as multiple deployments/services, we'll remove
  1135. // the '_' version. Not optimal, but takes care of the issue
  1136. func pruneDuplicateData(data map[string]map[string][]string) {
  1137. for _, podMap := range data {
  1138. for podName, values := range podMap {
  1139. podMap[podName] = pruneDuplicates(values)
  1140. }
  1141. }
  1142. }
  1143. // Determine if there is an underscore in the value of a slice. If so, replace _ with -, and then
  1144. // check to see if the result exists in the slice. If both are true, then we DO NOT include that
  1145. // original value in the new slice.
  1146. func pruneDuplicates(s []string) []string {
  1147. m := sliceToSet(s)
  1148. for _, v := range s {
  1149. if strings.Contains(v, "_") {
  1150. name := strings.Replace(v, "_", "-", -1)
  1151. if !m[name] {
  1152. m[name] = true
  1153. }
  1154. delete(m, v)
  1155. }
  1156. }
  1157. return setToSlice(m)
  1158. }
  1159. // Creates a map[string]bool containing the slice values as keys
  1160. func sliceToSet(s []string) map[string]bool {
  1161. m := make(map[string]bool)
  1162. for _, v := range s {
  1163. m[v] = true
  1164. }
  1165. return m
  1166. }
  1167. func setToSlice(m map[string]bool) []string {
  1168. var result []string
  1169. for k, _ := range m {
  1170. result = append(result, k)
  1171. }
  1172. return result
  1173. }
  1174. func costDataPassesFilters(costs *CostData, namespace string, cluster string) bool {
  1175. passesNamespace := namespace == "" || costs.Namespace == namespace
  1176. passesCluster := cluster == "" || costs.ClusterID == cluster
  1177. return passesNamespace && passesCluster
  1178. }
  1179. // Attempt to create a key for the request. Reduce the times to minutes in order to more easily group requests based on
  1180. // real time ranges. If for any reason, the key generation fails, return a uuid to ensure uniqueness.
  1181. func requestKeyFor(startString string, endString string, windowString string, filterNamespace string, filterCluster string, remoteEnabled bool) string {
  1182. fullLayout := "2006-01-02T15:04:05.000Z"
  1183. keyLayout := "2006-01-02T15:04Z"
  1184. sTime, err := time.Parse(fullLayout, startString)
  1185. if err != nil {
  1186. return uuid.New().String()
  1187. }
  1188. eTime, err := time.Parse(fullLayout, startString)
  1189. if err != nil {
  1190. return uuid.New().String()
  1191. }
  1192. startKey := sTime.Format(keyLayout)
  1193. endKey := eTime.Format(keyLayout)
  1194. return fmt.Sprintf("%s,%s,%s,%s,%s,%t", startKey, endKey, windowString, filterNamespace, filterCluster, remoteEnabled)
  1195. }
  1196. // Executes a range query for cost data
  1197. func (cm *CostModel) ComputeCostDataRange(cli prometheusClient.Client, clientset kubernetes.Interface, cp costAnalyzerCloud.Provider,
  1198. startString, endString, windowString string, filterNamespace string, filterCluster string, remoteEnabled bool) (map[string]*CostData, error) {
  1199. // Create a request key for request grouping. This key will be used to represent the cost-model result
  1200. // for the specific inputs to prevent multiple queries for identical data.
  1201. key := requestKeyFor(startString, endString, windowString, filterNamespace, filterCluster, remoteEnabled)
  1202. klog.V(4).Infof("ComputeCostDataRange with Key: %s", key)
  1203. // If there is already a request out that uses the same data, wait for it to return to share the results.
  1204. // Otherwise, start executing.
  1205. result, err, _ := cm.RequestGroup.Do(key, func() (interface{}, error) {
  1206. return cm.costDataRange(cli, clientset, cp, startString, endString, windowString, filterNamespace, filterCluster, remoteEnabled)
  1207. })
  1208. data, ok := result.(map[string]*CostData)
  1209. if !ok {
  1210. return nil, fmt.Errorf("Failed to cast result as map[string]*CostData")
  1211. }
  1212. return data, err
  1213. }
  1214. func (cm *CostModel) costDataRange(cli prometheusClient.Client, clientset kubernetes.Interface, cp costAnalyzerCloud.Provider,
  1215. startString, endString, windowString string, filterNamespace string, filterCluster string, remoteEnabled bool) (map[string]*CostData, error) {
  1216. queryRAMRequests := fmt.Sprintf(queryRAMRequestsStr, windowString, "", windowString, "")
  1217. queryRAMUsage := fmt.Sprintf(queryRAMUsageStr, windowString, "", windowString, "")
  1218. queryCPURequests := fmt.Sprintf(queryCPURequestsStr, windowString, "", windowString, "")
  1219. queryCPUUsage := fmt.Sprintf(queryCPUUsageStr, windowString, "")
  1220. queryRAMAlloc := fmt.Sprintf(queryRAMAllocation, windowString, "", windowString, "")
  1221. queryCPUAlloc := fmt.Sprintf(queryCPUAllocation, windowString, "", windowString, "")
  1222. queryGPURequests := fmt.Sprintf(queryGPURequestsStr, windowString, "", windowString, "", windowString, "")
  1223. queryPVRequests := fmt.Sprintf(queryPVRequestsStr)
  1224. queryNetZoneRequests := fmt.Sprintf(queryZoneNetworkUsage, windowString, "")
  1225. queryNetRegionRequests := fmt.Sprintf(queryRegionNetworkUsage, windowString, "")
  1226. queryNetInternetRequests := fmt.Sprintf(queryInternetNetworkUsage, windowString, "")
  1227. normalization := fmt.Sprintf(normalizationStr, windowString, "")
  1228. layout := "2006-01-02T15:04:05.000Z"
  1229. start, err := time.Parse(layout, startString)
  1230. if err != nil {
  1231. klog.V(1).Infof("Error parsing time " + startString + ". Error: " + err.Error())
  1232. return nil, err
  1233. }
  1234. end, err := time.Parse(layout, endString)
  1235. if err != nil {
  1236. klog.V(1).Infof("Error parsing time " + endString + ". Error: " + err.Error())
  1237. return nil, err
  1238. }
  1239. window, err := time.ParseDuration(windowString)
  1240. if err != nil {
  1241. klog.V(1).Infof("Error parsing time " + windowString + ". Error: " + err.Error())
  1242. return nil, err
  1243. }
  1244. clusterID := os.Getenv(clusterIDKey)
  1245. if remoteEnabled == true {
  1246. remoteLayout := "2006-01-02T15:04:05Z"
  1247. remoteStartStr := start.Format(remoteLayout)
  1248. remoteEndStr := end.Format(remoteLayout)
  1249. klog.V(1).Infof("Using remote database for query from %s to %s with window %s", startString, endString, windowString)
  1250. return CostDataRangeFromSQL("", "", windowString, remoteStartStr, remoteEndStr)
  1251. }
  1252. var wg sync.WaitGroup
  1253. wg.Add(20)
  1254. var promErr error
  1255. var resultRAMRequests interface{}
  1256. go func() {
  1257. defer wg.Done()
  1258. resultRAMRequests, promErr = QueryRange(cli, queryRAMRequests, start, end, window)
  1259. }()
  1260. var resultRAMUsage interface{}
  1261. go func() {
  1262. defer wg.Done()
  1263. resultRAMUsage, promErr = QueryRange(cli, queryRAMUsage, start, end, window)
  1264. }()
  1265. var resultCPURequests interface{}
  1266. go func() {
  1267. defer wg.Done()
  1268. resultCPURequests, promErr = QueryRange(cli, queryCPURequests, start, end, window)
  1269. }()
  1270. var resultCPUUsage interface{}
  1271. go func() {
  1272. defer wg.Done()
  1273. resultCPUUsage, promErr = QueryRange(cli, queryCPUUsage, start, end, window)
  1274. }()
  1275. var resultRAMAllocations interface{}
  1276. go func() {
  1277. defer wg.Done()
  1278. resultRAMAllocations, promErr = QueryRange(cli, queryRAMAlloc, start, end, window)
  1279. }()
  1280. var resultCPUAllocations interface{}
  1281. go func() {
  1282. defer wg.Done()
  1283. resultCPUAllocations, promErr = QueryRange(cli, queryCPUAlloc, start, end, window)
  1284. }()
  1285. var resultGPURequests interface{}
  1286. go func() {
  1287. defer wg.Done()
  1288. resultGPURequests, promErr = QueryRange(cli, queryGPURequests, start, end, window)
  1289. }()
  1290. var resultPVRequests interface{}
  1291. go func() {
  1292. defer wg.Done()
  1293. resultPVRequests, promErr = QueryRange(cli, queryPVRequests, start, end, window)
  1294. }()
  1295. var resultNetZoneRequests interface{}
  1296. go func() {
  1297. defer wg.Done()
  1298. resultNetZoneRequests, promErr = QueryRange(cli, queryNetZoneRequests, start, end, window)
  1299. }()
  1300. var resultNetRegionRequests interface{}
  1301. go func() {
  1302. defer wg.Done()
  1303. resultNetRegionRequests, promErr = QueryRange(cli, queryNetRegionRequests, start, end, window)
  1304. }()
  1305. var resultNetInternetRequests interface{}
  1306. go func() {
  1307. defer wg.Done()
  1308. resultNetInternetRequests, promErr = QueryRange(cli, queryNetInternetRequests, start, end, window)
  1309. }()
  1310. var pvPodAllocationResults interface{}
  1311. go func() {
  1312. defer wg.Done()
  1313. pvPodAllocationResults, promErr = QueryRange(cli, fmt.Sprintf(queryPVCAllocation, windowString), start, end, window)
  1314. }()
  1315. var pvCostResults interface{}
  1316. go func() {
  1317. defer wg.Done()
  1318. pvCostResults, promErr = QueryRange(cli, fmt.Sprintf(queryPVHourlyCost, windowString), start, end, window)
  1319. }()
  1320. var nsLabelsResults interface{}
  1321. go func() {
  1322. defer wg.Done()
  1323. nsLabelsResults, promErr = QueryRange(cli, fmt.Sprintf(queryNSLabels, windowString), start, end, window)
  1324. }()
  1325. var podLabelsResults interface{}
  1326. go func() {
  1327. defer wg.Done()
  1328. podLabelsResults, promErr = QueryRange(cli, fmt.Sprintf(queryPodLabels, windowString), start, end, window)
  1329. }()
  1330. var serviceLabelsResults interface{}
  1331. go func() {
  1332. defer wg.Done()
  1333. serviceLabelsResults, promErr = QueryRange(cli, fmt.Sprintf(queryServiceLabels, windowString), start, end, window)
  1334. }()
  1335. var deploymentLabelsResults interface{}
  1336. go func() {
  1337. defer wg.Done()
  1338. deploymentLabelsResults, promErr = QueryRange(cli, fmt.Sprintf(queryDeploymentLabels, windowString), start, end, window)
  1339. }()
  1340. var statefulsetLabelsResults interface{}
  1341. go func() {
  1342. defer wg.Done()
  1343. statefulsetLabelsResults, promErr = QueryRange(cli, fmt.Sprintf(queryStatefulsetLabels, windowString), start, end, window)
  1344. }()
  1345. var normalizationResults interface{}
  1346. go func() {
  1347. defer wg.Done()
  1348. normalizationResults, promErr = QueryRange(cli, normalization, start, end, window)
  1349. }()
  1350. podDeploymentsMapping := make(map[string]map[string][]string)
  1351. podStatefulsetsMapping := make(map[string]map[string][]string)
  1352. podServicesMapping := make(map[string]map[string][]string)
  1353. namespaceLabelsMapping := make(map[string]map[string]string)
  1354. podlist := cm.Cache.GetAllPods()
  1355. var k8sErr error
  1356. go func() {
  1357. defer wg.Done()
  1358. podDeploymentsMapping, k8sErr = getPodDeployments(cm.Cache, podlist, clusterID)
  1359. if k8sErr != nil {
  1360. return
  1361. }
  1362. podStatefulsetsMapping, k8sErr = getPodStatefulsets(cm.Cache, podlist, clusterID)
  1363. if k8sErr != nil {
  1364. return
  1365. }
  1366. podServicesMapping, k8sErr = getPodServices(cm.Cache, podlist, clusterID)
  1367. if k8sErr != nil {
  1368. return
  1369. }
  1370. namespaceLabelsMapping, k8sErr = getNamespaceLabels(cm.Cache, clusterID)
  1371. if k8sErr != nil {
  1372. return
  1373. }
  1374. }()
  1375. wg.Wait()
  1376. defer measureTime(time.Now(), "costDataRange: Processing Query Data")
  1377. if promErr != nil {
  1378. return nil, fmt.Errorf("Error querying prometheus: %s", promErr.Error())
  1379. }
  1380. if k8sErr != nil {
  1381. return nil, fmt.Errorf("Error querying the kubernetes api: %s", k8sErr.Error())
  1382. }
  1383. normalizationValue, err := getNormalizations(normalizationResults)
  1384. if err != nil {
  1385. return nil, fmt.Errorf("error computing normalization for start=%s, end=%s, window=%s: %s",
  1386. start, end, window, err.Error())
  1387. }
  1388. profileStart := time.Now()
  1389. nodes, err := cm.GetNodeCost(cp)
  1390. if err != nil {
  1391. klog.V(1).Infof("Warning, no cost model available: " + err.Error())
  1392. return nil, err
  1393. }
  1394. measureTime(profileStart, "GetNodeCost")
  1395. pvClaimMapping, err := GetPVInfo(resultPVRequests, clusterID)
  1396. if err != nil {
  1397. // Just log for compatibility with KSM less than 1.6
  1398. klog.Infof("Unable to get PV Data: %s", err.Error())
  1399. }
  1400. if pvClaimMapping != nil {
  1401. err = addPVData(cm.Cache, pvClaimMapping, cp)
  1402. if err != nil {
  1403. return nil, err
  1404. }
  1405. }
  1406. pvCostMapping, err := GetPVCostMetrics(pvCostResults, clusterID)
  1407. if err != nil {
  1408. klog.V(1).Infof("Unable to get PV Hourly Cost Data: %s", err.Error())
  1409. }
  1410. pvAllocationMapping, err := GetPVAllocationMetrics(pvPodAllocationResults, clusterID)
  1411. if err != nil {
  1412. klog.V(1).Infof("Unable to get PV Allocation Cost Data: %s", err.Error())
  1413. }
  1414. if pvAllocationMapping != nil {
  1415. addMetricPVData(pvAllocationMapping, pvCostMapping, cp)
  1416. }
  1417. nsLabels, err := GetNamespaceLabelsMetrics(nsLabelsResults, clusterID)
  1418. if err != nil {
  1419. klog.V(1).Infof("Unable to get Namespace Labels for Metrics: %s", err.Error())
  1420. }
  1421. if nsLabels != nil {
  1422. appendNamespaceLabels(namespaceLabelsMapping, nsLabels)
  1423. }
  1424. podLabels, err := GetPodLabelsMetrics(podLabelsResults, clusterID)
  1425. if err != nil {
  1426. klog.V(1).Infof("Unable to get Pod Labels for Metrics: %s", err.Error())
  1427. }
  1428. serviceLabels, err := GetServiceSelectorLabelsMetrics(serviceLabelsResults, clusterID)
  1429. if err != nil {
  1430. klog.V(1).Infof("Unable to get Service Selector Labels for Metrics: %s", err.Error())
  1431. }
  1432. deploymentLabels, err := GetDeploymentMatchLabelsMetrics(deploymentLabelsResults, clusterID)
  1433. if err != nil {
  1434. klog.V(1).Infof("Unable to get Deployment Match Labels for Metrics: %s", err.Error())
  1435. }
  1436. statefulsetLabels, err := GetStatefulsetMatchLabelsMetrics(statefulsetLabelsResults, clusterID)
  1437. if err != nil {
  1438. klog.V(1).Infof("Unable to get Deployment Match Labels for Metrics: %s", err.Error())
  1439. }
  1440. podStatefulsetMetricsMapping, err := getPodDeploymentsWithMetrics(statefulsetLabels, podLabels)
  1441. if err != nil {
  1442. klog.V(1).Infof("Unable to get match Statefulset Labels Metrics to Pods: %s", err.Error())
  1443. }
  1444. appendLabelsList(podStatefulsetsMapping, podStatefulsetMetricsMapping)
  1445. podDeploymentsMetricsMapping, err := getPodDeploymentsWithMetrics(deploymentLabels, podLabels)
  1446. if err != nil {
  1447. klog.V(1).Infof("Unable to get match Deployment Labels Metrics to Pods: %s", err.Error())
  1448. }
  1449. appendLabelsList(podDeploymentsMapping, podDeploymentsMetricsMapping)
  1450. podServicesMetricsMapping, err := getPodServicesWithMetrics(serviceLabels, podLabels)
  1451. if err != nil {
  1452. klog.V(1).Infof("Unable to get match Service Labels Metrics to Pods: %s", err.Error())
  1453. }
  1454. appendLabelsList(podServicesMapping, podServicesMetricsMapping)
  1455. networkUsageMap, err := GetNetworkUsageData(resultNetZoneRequests, resultNetRegionRequests, resultNetInternetRequests, clusterID)
  1456. if err != nil {
  1457. klog.V(1).Infof("Unable to get Network Cost Data: %s", err.Error())
  1458. networkUsageMap = make(map[string]*NetworkUsageData)
  1459. }
  1460. profileStart = time.Now()
  1461. containerNameCost := make(map[string]*CostData)
  1462. containers := make(map[string]bool)
  1463. otherClusterPVRecorded := make(map[string]bool)
  1464. RAMReqMap, err := GetContainerMetricVectors(resultRAMRequests, true, normalizationValue, clusterID)
  1465. if err != nil {
  1466. return nil, err
  1467. }
  1468. for key := range RAMReqMap {
  1469. containers[key] = true
  1470. }
  1471. RAMUsedMap, err := GetContainerMetricVectors(resultRAMUsage, true, normalizationValue, clusterID)
  1472. if err != nil {
  1473. return nil, err
  1474. }
  1475. for key := range RAMUsedMap {
  1476. containers[key] = true
  1477. }
  1478. CPUReqMap, err := GetContainerMetricVectors(resultCPURequests, true, normalizationValue, clusterID)
  1479. if err != nil {
  1480. return nil, err
  1481. }
  1482. for key := range CPUReqMap {
  1483. containers[key] = true
  1484. }
  1485. CPUUsedMap, err := GetContainerMetricVectors(resultCPUUsage, false, normalizationValue, clusterID) // No need to normalize here, as this comes from a counter
  1486. if err != nil {
  1487. return nil, err
  1488. }
  1489. for key := range CPUUsedMap {
  1490. containers[key] = true
  1491. }
  1492. RAMAllocMap, err := GetContainerMetricVectors(resultRAMAllocations, true, normalizationValue, clusterID)
  1493. if err != nil {
  1494. return nil, err
  1495. }
  1496. for key := range RAMAllocMap {
  1497. containers[key] = true
  1498. }
  1499. CPUAllocMap, err := GetContainerMetricVectors(resultCPUAllocations, true, normalizationValue, clusterID)
  1500. if err != nil {
  1501. return nil, err
  1502. }
  1503. for key := range CPUAllocMap {
  1504. containers[key] = true
  1505. }
  1506. GPUReqMap, err := GetContainerMetricVectors(resultGPURequests, true, normalizationValue, clusterID)
  1507. if err != nil {
  1508. return nil, err
  1509. }
  1510. for key := range GPUReqMap {
  1511. containers[key] = true
  1512. }
  1513. measureTime(profileStart, "GetContainerMetricVectors")
  1514. // Request metrics can show up after pod eviction and completion.
  1515. // This method synchronizes requests to allocations such that when
  1516. // allocation is 0, so are requests
  1517. applyAllocationToRequests(RAMAllocMap, RAMReqMap)
  1518. applyAllocationToRequests(CPUAllocMap, CPUReqMap)
  1519. currentContainers := make(map[string]v1.Pod)
  1520. for _, pod := range podlist {
  1521. if pod.Status.Phase != v1.PodRunning {
  1522. continue
  1523. }
  1524. cs, err := newContainerMetricsFromPod(*pod, clusterID)
  1525. if err != nil {
  1526. return nil, err
  1527. }
  1528. for _, c := range cs {
  1529. 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.
  1530. currentContainers[c.Key()] = *pod
  1531. }
  1532. }
  1533. profileStart = time.Now()
  1534. missingNodes := make(map[string]*costAnalyzerCloud.Node)
  1535. missingContainers := make(map[string]*CostData)
  1536. for key := range containers {
  1537. if _, ok := containerNameCost[key]; ok {
  1538. continue // because ordering is important for the allocation model (all PV's applied to the first), just dedupe if it's already been added.
  1539. }
  1540. if pod, ok := currentContainers[key]; ok {
  1541. podName := pod.GetObjectMeta().GetName()
  1542. ns := pod.GetObjectMeta().GetNamespace()
  1543. nodeName := pod.Spec.NodeName
  1544. var nodeData *costAnalyzerCloud.Node
  1545. if _, ok := nodes[nodeName]; ok {
  1546. nodeData = nodes[nodeName]
  1547. }
  1548. nsKey := ns + "," + clusterID
  1549. var podDeployments []string
  1550. if _, ok := podDeploymentsMapping[nsKey]; ok {
  1551. if ds, ok := podDeploymentsMapping[nsKey][pod.GetObjectMeta().GetName()]; ok {
  1552. podDeployments = ds
  1553. } else {
  1554. podDeployments = []string{}
  1555. }
  1556. }
  1557. var podStatefulSets []string
  1558. if _, ok := podStatefulsetsMapping[nsKey]; ok {
  1559. if ds, ok := podStatefulsetsMapping[nsKey][pod.GetObjectMeta().GetName()]; ok {
  1560. podStatefulSets = ds
  1561. } else {
  1562. podStatefulSets = []string{}
  1563. }
  1564. }
  1565. var podPVs []*PersistentVolumeClaimData
  1566. podClaims := pod.Spec.Volumes
  1567. for _, vol := range podClaims {
  1568. if vol.PersistentVolumeClaim != nil {
  1569. name := vol.PersistentVolumeClaim.ClaimName
  1570. if pvClaim, ok := pvClaimMapping[ns+","+name+","+clusterID]; ok {
  1571. podPVs = append(podPVs, pvClaim)
  1572. }
  1573. }
  1574. }
  1575. var podNetCosts []*Vector
  1576. if usage, ok := networkUsageMap[ns+","+podName+","+clusterID]; ok {
  1577. netCosts, err := GetNetworkCost(usage, cp)
  1578. if err != nil {
  1579. klog.V(3).Infof("Error pulling network costs: %s", err.Error())
  1580. } else {
  1581. podNetCosts = netCosts
  1582. }
  1583. }
  1584. var podServices []string
  1585. if _, ok := podServicesMapping[nsKey]; ok {
  1586. if svcs, ok := podServicesMapping[nsKey][pod.GetObjectMeta().GetName()]; ok {
  1587. podServices = svcs
  1588. } else {
  1589. podServices = []string{}
  1590. }
  1591. }
  1592. nsLabels := namespaceLabelsMapping[nsKey]
  1593. podLabels := pod.GetObjectMeta().GetLabels()
  1594. if podLabels == nil {
  1595. podLabels = make(map[string]string)
  1596. }
  1597. for k, v := range nsLabels {
  1598. podLabels[k] = v
  1599. }
  1600. for i, container := range pod.Spec.Containers {
  1601. containerName := container.Name
  1602. newKey := newContainerMetricFromValues(ns, podName, containerName, pod.Spec.NodeName, clusterID).Key()
  1603. RAMReqV, ok := RAMReqMap[newKey]
  1604. if !ok {
  1605. klog.V(4).Info("no RAM requests for " + newKey)
  1606. RAMReqV = []*Vector{}
  1607. }
  1608. RAMUsedV, ok := RAMUsedMap[newKey]
  1609. if !ok {
  1610. klog.V(4).Info("no RAM usage for " + newKey)
  1611. RAMUsedV = []*Vector{}
  1612. }
  1613. CPUReqV, ok := CPUReqMap[newKey]
  1614. if !ok {
  1615. klog.V(4).Info("no CPU requests for " + newKey)
  1616. CPUReqV = []*Vector{}
  1617. }
  1618. CPUUsedV, ok := CPUUsedMap[newKey]
  1619. if !ok {
  1620. klog.V(4).Info("no CPU usage for " + newKey)
  1621. CPUUsedV = []*Vector{}
  1622. }
  1623. RAMAllocsV, ok := RAMAllocMap[newKey]
  1624. if !ok {
  1625. klog.V(4).Info("no RAM allocation for " + newKey)
  1626. RAMAllocsV = []*Vector{}
  1627. }
  1628. CPUAllocsV, ok := CPUAllocMap[newKey]
  1629. if !ok {
  1630. klog.V(4).Info("no CPU allocation for " + newKey)
  1631. CPUAllocsV = []*Vector{}
  1632. }
  1633. GPUReqV, ok := GPUReqMap[newKey]
  1634. if !ok {
  1635. klog.V(4).Info("no GPU requests for " + newKey)
  1636. GPUReqV = []*Vector{}
  1637. }
  1638. var pvReq []*PersistentVolumeClaimData
  1639. var netReq []*Vector
  1640. if i == 0 { // avoid duplicating by just assigning all claims to the first container.
  1641. pvReq = podPVs
  1642. netReq = podNetCosts
  1643. }
  1644. costs := &CostData{
  1645. Name: containerName,
  1646. PodName: podName,
  1647. NodeName: nodeName,
  1648. Namespace: ns,
  1649. Deployments: podDeployments,
  1650. Services: podServices,
  1651. Daemonsets: getDaemonsetsOfPod(pod),
  1652. Jobs: getJobsOfPod(pod),
  1653. Statefulsets: podStatefulSets,
  1654. NodeData: nodeData,
  1655. RAMReq: RAMReqV,
  1656. RAMUsed: RAMUsedV,
  1657. CPUReq: CPUReqV,
  1658. CPUUsed: CPUUsedV,
  1659. RAMAllocation: RAMAllocsV,
  1660. CPUAllocation: CPUAllocsV,
  1661. GPUReq: GPUReqV,
  1662. PVCData: pvReq,
  1663. Labels: podLabels,
  1664. NetworkData: netReq,
  1665. NamespaceLabels: nsLabels,
  1666. ClusterID: clusterID,
  1667. }
  1668. if costDataPassesFilters(costs, filterNamespace, filterCluster) {
  1669. containerNameCost[newKey] = costs
  1670. }
  1671. }
  1672. } else {
  1673. // The container has been deleted, or is from a different clusterID
  1674. // Not all information is sent to prometheus via ksm, so fill out what we can without k8s api
  1675. klog.V(4).Info("The container " + key + " has been deleted. Calculating allocation but resulting object will be missing data.")
  1676. c, _ := NewContainerMetricFromKey(key)
  1677. RAMReqV, ok := RAMReqMap[key]
  1678. if !ok {
  1679. klog.V(4).Info("no RAM requests for " + key)
  1680. RAMReqV = []*Vector{}
  1681. }
  1682. RAMUsedV, ok := RAMUsedMap[key]
  1683. if !ok {
  1684. klog.V(4).Info("no RAM usage for " + key)
  1685. RAMUsedV = []*Vector{}
  1686. }
  1687. CPUReqV, ok := CPUReqMap[key]
  1688. if !ok {
  1689. klog.V(4).Info("no CPU requests for " + key)
  1690. CPUReqV = []*Vector{}
  1691. }
  1692. CPUUsedV, ok := CPUUsedMap[key]
  1693. if !ok {
  1694. klog.V(4).Info("no CPU usage for " + key)
  1695. CPUUsedV = []*Vector{}
  1696. }
  1697. RAMAllocsV, ok := RAMAllocMap[key]
  1698. if !ok {
  1699. klog.V(4).Info("no RAM allocation for " + key)
  1700. RAMAllocsV = []*Vector{}
  1701. }
  1702. CPUAllocsV, ok := CPUAllocMap[key]
  1703. if !ok {
  1704. klog.V(4).Info("no CPU allocation for " + key)
  1705. CPUAllocsV = []*Vector{}
  1706. }
  1707. GPUReqV, ok := GPUReqMap[key]
  1708. if !ok {
  1709. klog.V(4).Info("no GPU requests for " + key)
  1710. GPUReqV = []*Vector{}
  1711. }
  1712. node, ok := nodes[c.NodeName]
  1713. if !ok {
  1714. klog.V(4).Infof("Node \"%s\" has been deleted from Kubernetes. Query historical data to get it.", c.NodeName)
  1715. if n, ok := missingNodes[c.NodeName]; ok {
  1716. node = n
  1717. } else {
  1718. node = &costAnalyzerCloud.Node{}
  1719. missingNodes[c.NodeName] = node
  1720. }
  1721. }
  1722. nsKey := c.Namespace + "," + c.ClusterID
  1723. podKey := c.Namespace + "," + c.PodName + "," + c.ClusterID
  1724. namespaceLabels, ok := namespaceLabelsMapping[nsKey]
  1725. if !ok {
  1726. klog.V(3).Infof("Missing data for namespace %s", c.Namespace)
  1727. }
  1728. pLabels := podLabels[podKey]
  1729. if pLabels == nil {
  1730. pLabels = make(map[string]string)
  1731. }
  1732. for k, v := range namespaceLabels {
  1733. pLabels[k] = v
  1734. }
  1735. var podDeployments []string
  1736. if _, ok := podDeploymentsMapping[nsKey]; ok {
  1737. if ds, ok := podDeploymentsMapping[nsKey][c.PodName]; ok {
  1738. podDeployments = ds
  1739. } else {
  1740. podDeployments = []string{}
  1741. }
  1742. }
  1743. var podServices []string
  1744. if _, ok := podServicesMapping[nsKey]; ok {
  1745. if svcs, ok := podServicesMapping[nsKey][c.PodName]; ok {
  1746. podServices = svcs
  1747. } else {
  1748. podServices = []string{}
  1749. }
  1750. }
  1751. var podPVs []*PersistentVolumeClaimData
  1752. var podNetCosts []*Vector
  1753. // For PVC data, we'll need to find the claim mapping and cost data. Will need to append
  1754. // cost data since that was populated by cluster data previously. We do this with
  1755. // the pod_pvc_allocation metric
  1756. podPVData, ok := pvAllocationMapping[podKey]
  1757. if !ok {
  1758. klog.V(4).Infof("Failed to locate pv allocation mapping for missing pod.")
  1759. }
  1760. // For network costs, we'll use existing map since it should still contain the
  1761. // correct data.
  1762. var podNetworkCosts []*Vector
  1763. if usage, ok := networkUsageMap[podKey]; ok {
  1764. netCosts, err := GetNetworkCost(usage, cp)
  1765. if err != nil {
  1766. klog.V(3).Infof("Error pulling network costs: %s", err.Error())
  1767. } else {
  1768. podNetworkCosts = netCosts
  1769. }
  1770. }
  1771. // Check to see if any other data has been recorded for this namespace, pod, clusterId
  1772. // Follow the pattern of only allowing claims data per pod
  1773. if !otherClusterPVRecorded[podKey] {
  1774. otherClusterPVRecorded[podKey] = true
  1775. podPVs = podPVData
  1776. podNetCosts = podNetworkCosts
  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. RAMReq: RAMReqV,
  1787. RAMUsed: RAMUsedV,
  1788. CPUReq: CPUReqV,
  1789. CPUUsed: CPUUsedV,
  1790. RAMAllocation: RAMAllocsV,
  1791. CPUAllocation: CPUAllocsV,
  1792. GPUReq: GPUReqV,
  1793. Labels: pLabels,
  1794. NamespaceLabels: namespaceLabels,
  1795. PVCData: podPVs,
  1796. NetworkData: podNetCosts,
  1797. ClusterID: c.ClusterID,
  1798. }
  1799. if costDataPassesFilters(costs, filterNamespace, filterCluster) {
  1800. containerNameCost[key] = costs
  1801. missingContainers[key] = costs
  1802. }
  1803. }
  1804. }
  1805. measureTime(profileStart, "Build CostData map")
  1806. w := end.Sub(start)
  1807. w += window
  1808. if w.Minutes() > 0 {
  1809. wStr := fmt.Sprintf("%dm", int(w.Minutes()))
  1810. err = findDeletedNodeInfo(cli, missingNodes, wStr)
  1811. if err != nil {
  1812. klog.V(1).Infof("Error fetching historical node data: %s", err.Error())
  1813. }
  1814. }
  1815. return containerNameCost, err
  1816. }
  1817. func applyAllocationToRequests(allocationMap map[string][]*Vector, requestMap map[string][]*Vector) {
  1818. // The result of the normalize operation will be a new []*Vector to replace the requests
  1819. normalizeOp := func(r *Vector, x *float64, y *float64) bool {
  1820. // Omit data (return false) if both x and y inputs don't exist
  1821. if x == nil || y == nil {
  1822. return false
  1823. }
  1824. // If the allocation value is 0, 0 out request value
  1825. if *x == 0 {
  1826. r.Value = 0
  1827. } else {
  1828. r.Value = *y
  1829. }
  1830. return true
  1831. }
  1832. // Run normalization on all request vectors in the mapping
  1833. for k, requests := range requestMap {
  1834. // Only run normalization where there are valid allocations
  1835. allocations, ok := allocationMap[k]
  1836. if !ok {
  1837. delete(requestMap, k)
  1838. continue
  1839. }
  1840. // Replace request map with normalized
  1841. requestMap[k] = ApplyVectorOp(allocations, requests, normalizeOp)
  1842. }
  1843. }
  1844. func addMetricPVData(pvAllocationMap map[string][]*PersistentVolumeClaimData, pvCostMap map[string]*costAnalyzerCloud.PV, cp costAnalyzerCloud.Provider) {
  1845. cfg, err := cp.GetConfig()
  1846. if err != nil {
  1847. klog.V(1).Infof("Failed to get provider config while adding pv metrics data.")
  1848. return
  1849. }
  1850. for _, pvcDataArray := range pvAllocationMap {
  1851. for _, pvcData := range pvcDataArray {
  1852. costKey := fmt.Sprintf("%s,%s", pvcData.VolumeName, pvcData.ClusterID)
  1853. pvCost, ok := pvCostMap[costKey]
  1854. if !ok {
  1855. pvcData.Volume = &costAnalyzerCloud.PV{
  1856. Cost: cfg.Storage,
  1857. }
  1858. continue
  1859. }
  1860. pvcData.Volume = pvCost
  1861. }
  1862. }
  1863. }
  1864. // Append labels into nsLabels iff the ns key doesn't already exist
  1865. func appendNamespaceLabels(nsLabels map[string]map[string]string, labels map[string]map[string]string) {
  1866. for k, v := range labels {
  1867. if _, ok := nsLabels[k]; !ok {
  1868. nsLabels[k] = v
  1869. }
  1870. }
  1871. }
  1872. func appendLabelsList(mainLabels map[string]map[string][]string, labels map[string]map[string][]string) {
  1873. for k, v := range labels {
  1874. mainLabels[k] = v
  1875. }
  1876. }
  1877. func getNamespaceLabels(cache clustercache.ClusterCache, clusterID string) (map[string]map[string]string, error) {
  1878. nsToLabels := make(map[string]map[string]string)
  1879. nss := cache.GetAllNamespaces()
  1880. for _, ns := range nss {
  1881. nsToLabels[ns.Name+","+clusterID] = ns.Labels
  1882. }
  1883. return nsToLabels, nil
  1884. }
  1885. func getDaemonsetsOfPod(pod v1.Pod) []string {
  1886. for _, ownerReference := range pod.ObjectMeta.OwnerReferences {
  1887. if ownerReference.Kind == "DaemonSet" {
  1888. return []string{ownerReference.Name}
  1889. }
  1890. }
  1891. return []string{}
  1892. }
  1893. func getJobsOfPod(pod v1.Pod) []string {
  1894. for _, ownerReference := range pod.ObjectMeta.OwnerReferences {
  1895. if ownerReference.Kind == "Job" {
  1896. return []string{ownerReference.Name}
  1897. }
  1898. }
  1899. return []string{}
  1900. }
  1901. func getStatefulSetsOfPod(pod v1.Pod) []string {
  1902. for _, ownerReference := range pod.ObjectMeta.OwnerReferences {
  1903. if ownerReference.Kind == "StatefulSet" {
  1904. return []string{ownerReference.Name}
  1905. }
  1906. }
  1907. return []string{}
  1908. }
  1909. type PersistentVolumeClaimData struct {
  1910. Class string `json:"class"`
  1911. Claim string `json:"claim"`
  1912. Namespace string `json:"namespace"`
  1913. ClusterID string `json:"clusterId"`
  1914. VolumeName string `json:"volumeName"`
  1915. Volume *costAnalyzerCloud.PV `json:"persistentVolume"`
  1916. Values []*Vector `json:"values"`
  1917. }
  1918. func getCost(qr interface{}) (map[string][]*Vector, error) {
  1919. toReturn := make(map[string][]*Vector)
  1920. result, err := NewQueryResults(qr)
  1921. if err != nil {
  1922. return toReturn, err
  1923. }
  1924. for _, val := range result {
  1925. instance, err := val.GetString("instance")
  1926. if err != nil {
  1927. return toReturn, err
  1928. }
  1929. toReturn[instance] = val.Values
  1930. }
  1931. return toReturn, nil
  1932. }
  1933. func QueryRange(cli prometheusClient.Client, query string, start, end time.Time, step time.Duration) (interface{}, error) {
  1934. u := cli.URL(epQueryRange, nil)
  1935. q := u.Query()
  1936. q.Set("query", query)
  1937. q.Set("start", start.Format(time.RFC3339Nano))
  1938. q.Set("end", end.Format(time.RFC3339Nano))
  1939. q.Set("step", strconv.FormatFloat(step.Seconds(), 'f', 3, 64))
  1940. u.RawQuery = q.Encode()
  1941. req, err := http.NewRequest(http.MethodPost, u.String(), nil)
  1942. if err != nil {
  1943. return nil, err
  1944. }
  1945. _, body, warnings, err := cli.Do(context.Background(), req)
  1946. for _, w := range warnings {
  1947. klog.V(3).Infof("%s", w)
  1948. }
  1949. if err != nil {
  1950. return nil, fmt.Errorf("Error %s fetching query %s", err.Error(), query)
  1951. }
  1952. var toReturn interface{}
  1953. err = json.Unmarshal(body, &toReturn)
  1954. if err != nil {
  1955. return nil, fmt.Errorf("Error %s fetching query %s", err.Error(), query)
  1956. }
  1957. return toReturn, err
  1958. }
  1959. func Query(cli prometheusClient.Client, query string) (interface{}, error) {
  1960. u := cli.URL(epQuery, nil)
  1961. q := u.Query()
  1962. q.Set("query", query)
  1963. u.RawQuery = q.Encode()
  1964. req, err := http.NewRequest(http.MethodPost, u.String(), nil)
  1965. if err != nil {
  1966. return nil, err
  1967. }
  1968. resp, body, warnings, err := cli.Do(context.Background(), req)
  1969. for _, w := range warnings {
  1970. klog.V(3).Infof("%s", w)
  1971. }
  1972. if err != nil {
  1973. if resp == nil {
  1974. return nil, fmt.Errorf("Error %s fetching query %s", err.Error(), query)
  1975. }
  1976. return nil, fmt.Errorf("%d Error %s fetching query %s", resp.StatusCode, err.Error(), query)
  1977. }
  1978. var toReturn interface{}
  1979. err = json.Unmarshal(body, &toReturn)
  1980. if err != nil {
  1981. return nil, fmt.Errorf("Error %s fetching query %s", err.Error(), query)
  1982. }
  1983. return toReturn, nil
  1984. }
  1985. //todo: don't cast, implement unmarshaler interface
  1986. func getNormalization(qr interface{}) (float64, error) {
  1987. queryResults, err := NewQueryResults(qr)
  1988. if err != nil {
  1989. return 0, err
  1990. }
  1991. if len(queryResults) > 0 {
  1992. values := queryResults[0].Values
  1993. if len(values) > 0 {
  1994. return values[0].Value, nil
  1995. }
  1996. return 0, fmt.Errorf("Improperly formatted datapoint from Prometheus")
  1997. }
  1998. return 0, fmt.Errorf("Normalization data is empty, kube-state-metrics or node-exporter may not be running")
  1999. }
  2000. //todo: don't cast, implement unmarshaler interface
  2001. func getNormalizations(qr interface{}) ([]*Vector, error) {
  2002. queryResults, err := NewQueryResults(qr)
  2003. if err != nil {
  2004. return nil, err
  2005. }
  2006. if len(queryResults) > 0 {
  2007. vectors := []*Vector{}
  2008. for _, value := range queryResults {
  2009. vectors = append(vectors, value.Values...)
  2010. }
  2011. return vectors, nil
  2012. }
  2013. return nil, fmt.Errorf("normalization data is empty: time window may be invalid or kube-state-metrics or node-exporter may not be running")
  2014. }
  2015. type ContainerMetric struct {
  2016. Namespace string
  2017. PodName string
  2018. ContainerName string
  2019. NodeName string
  2020. ClusterID string
  2021. }
  2022. func (c *ContainerMetric) Key() string {
  2023. return c.Namespace + "," + c.PodName + "," + c.ContainerName + "," + c.NodeName + "," + c.ClusterID
  2024. }
  2025. func NewContainerMetricFromKey(key string) (*ContainerMetric, error) {
  2026. s := strings.Split(key, ",")
  2027. if len(s) == 5 {
  2028. return &ContainerMetric{
  2029. Namespace: s[0],
  2030. PodName: s[1],
  2031. ContainerName: s[2],
  2032. NodeName: s[3],
  2033. ClusterID: s[4],
  2034. }, nil
  2035. }
  2036. return nil, fmt.Errorf("Not a valid key")
  2037. }
  2038. func newContainerMetricFromValues(ns string, podName string, containerName string, nodeName string, clusterId string) *ContainerMetric {
  2039. return &ContainerMetric{
  2040. Namespace: ns,
  2041. PodName: podName,
  2042. ContainerName: containerName,
  2043. NodeName: nodeName,
  2044. ClusterID: clusterId,
  2045. }
  2046. }
  2047. func newContainerMetricsFromPod(pod v1.Pod, clusterID string) ([]*ContainerMetric, error) {
  2048. podName := pod.GetObjectMeta().GetName()
  2049. ns := pod.GetObjectMeta().GetNamespace()
  2050. node := pod.Spec.NodeName
  2051. var cs []*ContainerMetric
  2052. for _, container := range pod.Spec.Containers {
  2053. containerName := container.Name
  2054. cs = append(cs, &ContainerMetric{
  2055. Namespace: ns,
  2056. PodName: podName,
  2057. ContainerName: containerName,
  2058. NodeName: node,
  2059. ClusterID: clusterID,
  2060. })
  2061. }
  2062. return cs, nil
  2063. }
  2064. func newContainerMetricFromPrometheus(metrics map[string]interface{}, defaultClusterID string) (*ContainerMetric, error) {
  2065. cName, ok := metrics["container_name"]
  2066. if !ok {
  2067. return nil, fmt.Errorf("Prometheus vector does not have container name")
  2068. }
  2069. containerName, ok := cName.(string)
  2070. if !ok {
  2071. return nil, fmt.Errorf("Prometheus vector does not have string container name")
  2072. }
  2073. pName, ok := metrics["pod_name"]
  2074. if !ok {
  2075. return nil, fmt.Errorf("Prometheus vector does not have pod name")
  2076. }
  2077. podName, ok := pName.(string)
  2078. if !ok {
  2079. return nil, fmt.Errorf("Prometheus vector does not have string pod name")
  2080. }
  2081. ns, ok := metrics["namespace"]
  2082. if !ok {
  2083. return nil, fmt.Errorf("Prometheus vector does not have namespace")
  2084. }
  2085. namespace, ok := ns.(string)
  2086. if !ok {
  2087. return nil, fmt.Errorf("Prometheus vector does not have string namespace")
  2088. }
  2089. node, ok := metrics["node"]
  2090. if !ok {
  2091. klog.V(4).Info("Prometheus vector does not have node name")
  2092. node = ""
  2093. }
  2094. nodeName, ok := node.(string)
  2095. if !ok {
  2096. return nil, fmt.Errorf("Prometheus vector does not have string node")
  2097. }
  2098. cid, ok := metrics["cluster_id"]
  2099. if !ok {
  2100. klog.V(4).Info("Prometheus vector does not have cluster id")
  2101. cid = defaultClusterID
  2102. }
  2103. clusterID, ok := cid.(string)
  2104. if !ok {
  2105. return nil, fmt.Errorf("Prometheus vector does not have string cluster_id")
  2106. }
  2107. return &ContainerMetric{
  2108. ContainerName: containerName,
  2109. PodName: podName,
  2110. Namespace: namespace,
  2111. NodeName: nodeName,
  2112. ClusterID: clusterID,
  2113. }, nil
  2114. }
  2115. type KeyTuple struct {
  2116. Namespace string
  2117. Key string
  2118. ClusterID string
  2119. }
  2120. func NewKeyTuple(key string) (*KeyTuple, error) {
  2121. r := strings.Split(key, ",")
  2122. if len(r) != 3 {
  2123. return nil, fmt.Errorf("NewKeyTuple() Provided key not containing exactly 3 components.")
  2124. }
  2125. return &KeyTuple{
  2126. Namespace: r[0],
  2127. Key: r[1],
  2128. ClusterID: r[2],
  2129. }, nil
  2130. }
  2131. func GetContainerMetricVector(qr interface{}, normalize bool, normalizationValue float64, defaultClusterID string) (map[string][]*Vector, error) {
  2132. result, err := NewQueryResults(qr)
  2133. if err != nil {
  2134. return nil, err
  2135. }
  2136. containerData := make(map[string][]*Vector)
  2137. for _, val := range result {
  2138. containerMetric, err := newContainerMetricFromPrometheus(val.Metric, defaultClusterID)
  2139. if err != nil {
  2140. return nil, err
  2141. }
  2142. if normalize && normalizationValue != 0 {
  2143. for _, v := range val.Values {
  2144. v.Value = v.Value / normalizationValue
  2145. }
  2146. }
  2147. containerData[containerMetric.Key()] = val.Values
  2148. }
  2149. return containerData, nil
  2150. }
  2151. func GetContainerMetricVectors(qr interface{}, normalize bool, normalizationValues []*Vector, defaultClusterID string) (map[string][]*Vector, error) {
  2152. result, err := NewQueryResults(qr)
  2153. if err != nil {
  2154. return nil, err
  2155. }
  2156. containerData := make(map[string][]*Vector)
  2157. for _, val := range result {
  2158. containerMetric, err := newContainerMetricFromPrometheus(val.Metric, defaultClusterID)
  2159. if err != nil {
  2160. return nil, err
  2161. }
  2162. normalizedVectors := NormalizeVectorByVector(val.Values, normalizationValues)
  2163. containerData[containerMetric.Key()] = normalizedVectors
  2164. }
  2165. return containerData, nil
  2166. }
  2167. func wrapPrometheusError(qr interface{}) (string, error) {
  2168. e, ok := qr.(map[string]interface{})["error"]
  2169. if !ok {
  2170. return "", fmt.Errorf("Unexpected response from Prometheus")
  2171. }
  2172. eStr, ok := e.(string)
  2173. return eStr, nil
  2174. }
  2175. func measureTime(start time.Time, name string) {
  2176. elapsed := time.Since(start)
  2177. klog.V(3).Infof("[Profiler] %s took %s", name, elapsed)
  2178. }