costmodel.go 66 KB

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