main.go 26 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772
  1. package main
  2. import (
  3. "context"
  4. "encoding/json"
  5. "flag"
  6. "fmt"
  7. "net"
  8. "net/http"
  9. "os"
  10. "reflect"
  11. "strconv"
  12. "strings"
  13. "time"
  14. "k8s.io/klog"
  15. "github.com/julienschmidt/httprouter"
  16. costAnalyzerCloud "github.com/kubecost/cost-model/cloud"
  17. costModel "github.com/kubecost/cost-model/costmodel"
  18. prometheusClient "github.com/prometheus/client_golang/api"
  19. prometheusAPI "github.com/prometheus/client_golang/api/prometheus/v1"
  20. v1 "k8s.io/api/core/v1"
  21. metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
  22. "github.com/prometheus/client_golang/prometheus"
  23. "github.com/prometheus/client_golang/prometheus/promhttp"
  24. "k8s.io/apimachinery/pkg/fields"
  25. "k8s.io/client-go/kubernetes"
  26. "k8s.io/client-go/rest"
  27. "k8s.io/client-go/tools/cache"
  28. )
  29. const (
  30. prometheusServerEndpointEnvVar = "PROMETHEUS_SERVER_ENDPOINT"
  31. prometheusTroubleshootingEp = "http://docs.kubecost.com/custom-prom#troubleshoot"
  32. remoteEnabled = "REMOTE_WRITE_ENABLED"
  33. )
  34. var (
  35. // gitCommit is set by the build system
  36. gitCommit string
  37. )
  38. type Accesses struct {
  39. PrometheusClient prometheusClient.Client
  40. KubeClientSet kubernetes.Interface
  41. Cloud costAnalyzerCloud.Provider
  42. CPUPriceRecorder *prometheus.GaugeVec
  43. RAMPriceRecorder *prometheus.GaugeVec
  44. PersistentVolumePriceRecorder *prometheus.GaugeVec
  45. GPUPriceRecorder *prometheus.GaugeVec
  46. NodeTotalPriceRecorder *prometheus.GaugeVec
  47. RAMAllocationRecorder *prometheus.GaugeVec
  48. CPUAllocationRecorder *prometheus.GaugeVec
  49. GPUAllocationRecorder *prometheus.GaugeVec
  50. ContainerUptimeRecorder *prometheus.GaugeVec
  51. ServiceSelectorRecorder *prometheus.GaugeVec
  52. DeploymentSelectorRecorder *prometheus.GaugeVec
  53. Model *costModel.CostModel
  54. }
  55. type DataEnvelope struct {
  56. Code int `json:"code"`
  57. Status string `json:"status"`
  58. Data interface{} `json:"data"`
  59. Message string `json:"message,omitempty"`
  60. }
  61. func wrapData(data interface{}, err error) []byte {
  62. var resp []byte
  63. if err != nil {
  64. klog.V(1).Infof("Error returned to client: %s", err.Error())
  65. resp, _ = json.Marshal(&DataEnvelope{
  66. Code: 500,
  67. Status: "error",
  68. Message: err.Error(),
  69. Data: data,
  70. })
  71. } else {
  72. resp, _ = json.Marshal(&DataEnvelope{
  73. Code: 200,
  74. Status: "success",
  75. Data: data,
  76. })
  77. }
  78. return resp
  79. }
  80. // RefreshPricingData needs to be called when a new node joins the fleet, since we cache the relevant subsets of pricing data to avoid storing the whole thing.
  81. func (a *Accesses) RefreshPricingData(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  82. w.Header().Set("Content-Type", "application/json")
  83. w.Header().Set("Access-Control-Allow-Origin", "*")
  84. err := a.Cloud.DownloadPricingData()
  85. w.Write(wrapData(nil, err))
  86. }
  87. func filterFields(fields string, data map[string]*costModel.CostData) map[string]costModel.CostData {
  88. fs := strings.Split(fields, ",")
  89. fmap := make(map[string]bool)
  90. for _, f := range fs {
  91. fieldNameLower := strings.ToLower(f) // convert to go struct name by uppercasing first letter
  92. klog.V(1).Infof("to delete: %s", fieldNameLower)
  93. fmap[fieldNameLower] = true
  94. }
  95. filteredData := make(map[string]costModel.CostData)
  96. for cname, costdata := range data {
  97. s := reflect.TypeOf(*costdata)
  98. val := reflect.ValueOf(*costdata)
  99. costdata2 := costModel.CostData{}
  100. cd2 := reflect.New(reflect.Indirect(reflect.ValueOf(costdata2)).Type()).Elem()
  101. n := s.NumField()
  102. for i := 0; i < n; i++ {
  103. field := s.Field(i)
  104. value := val.Field(i)
  105. value2 := cd2.Field(i)
  106. if _, ok := fmap[strings.ToLower(field.Name)]; !ok {
  107. value2.Set(reflect.Value(value))
  108. }
  109. }
  110. filteredData[cname] = cd2.Interface().(costModel.CostData)
  111. }
  112. return filteredData
  113. }
  114. func (a *Accesses) CostDataModel(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  115. w.Header().Set("Content-Type", "application/json")
  116. w.Header().Set("Access-Control-Allow-Origin", "*")
  117. window := r.URL.Query().Get("timeWindow")
  118. offset := r.URL.Query().Get("offset")
  119. fields := r.URL.Query().Get("filterFields")
  120. namespace := r.URL.Query().Get("namespace")
  121. aggregation := r.URL.Query().Get("aggregation")
  122. aggregationSubField := r.URL.Query().Get("aggregationSubfield")
  123. if offset != "" {
  124. offset = "offset " + offset
  125. }
  126. data, err := a.Model.ComputeCostData(a.PrometheusClient, a.KubeClientSet, a.Cloud, window, offset, namespace)
  127. if aggregation != "" {
  128. agg := costModel.AggregateCostModel(data, aggregation, aggregationSubField)
  129. w.Write(wrapData(agg, nil))
  130. } else {
  131. if fields != "" {
  132. filteredData := filterFields(fields, data)
  133. w.Write(wrapData(filteredData, err))
  134. } else {
  135. w.Write(wrapData(data, err))
  136. }
  137. }
  138. }
  139. func (a *Accesses) ClusterCosts(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  140. w.Header().Set("Content-Type", "application/json")
  141. w.Header().Set("Access-Control-Allow-Origin", "*")
  142. window := r.URL.Query().Get("window")
  143. offset := r.URL.Query().Get("offset")
  144. if offset != "" {
  145. offset = "offset " + offset
  146. }
  147. data, err := costModel.ClusterCosts(a.PrometheusClient, a.Cloud, window, offset)
  148. w.Write(wrapData(data, err))
  149. }
  150. func (a *Accesses) ClusterCostsOverTime(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  151. w.Header().Set("Content-Type", "application/json")
  152. w.Header().Set("Access-Control-Allow-Origin", "*")
  153. start := r.URL.Query().Get("start")
  154. end := r.URL.Query().Get("end")
  155. window := r.URL.Query().Get("window")
  156. offset := r.URL.Query().Get("offset")
  157. if offset != "" {
  158. offset = "offset " + offset
  159. }
  160. data, err := costModel.ClusterCostsOverTime(a.PrometheusClient, a.Cloud, start, end, window, offset)
  161. w.Write(wrapData(data, err))
  162. }
  163. func (a *Accesses) AggregateCostModel(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  164. w.Header().Set("Content-Type", "application/json")
  165. w.Header().Set("Access-Control-Allow-Origin", "*")
  166. window := r.URL.Query().Get("window")
  167. offset := r.URL.Query().Get("offset")
  168. aggregation := r.URL.Query().Get("aggregation")
  169. namespace := r.URL.Query().Get("namespace")
  170. aggregationSubField := r.URL.Query().Get("aggregationSubfield")
  171. endTime := time.Now()
  172. if offset != "" {
  173. o, err := time.ParseDuration(offset)
  174. if err != nil {
  175. w.Write(wrapData(nil, err))
  176. return
  177. }
  178. endTime = endTime.Add(-1 * o)
  179. }
  180. if window[len(window)-1:] == "d" {
  181. count := window[:len(window)-1]
  182. val, err := strconv.ParseInt(count, 10, 64)
  183. if err != nil {
  184. w.Write(wrapData(nil, err))
  185. return
  186. }
  187. val = val * 24
  188. window = fmt.Sprintf("%dh", val)
  189. }
  190. d, err := time.ParseDuration(window)
  191. if err != nil {
  192. w.Write(wrapData(nil, err))
  193. return
  194. }
  195. startTime := endTime.Add(-1 * d)
  196. layout := "2006-01-02T15:04:05.000Z"
  197. start := startTime.Format(layout)
  198. end := endTime.Format(layout)
  199. data, err := a.Model.ComputeCostDataRange(a.PrometheusClient, a.KubeClientSet, a.Cloud, start, end, "1h", namespace)
  200. if err != nil {
  201. w.Write(wrapData(nil, err))
  202. return
  203. }
  204. if aggregation != "" {
  205. agg := costModel.AggregateCostModel(data, aggregation, aggregationSubField)
  206. w.Write(wrapData(agg, nil))
  207. }
  208. }
  209. func (a *Accesses) CostDataModelRange(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  210. w.Header().Set("Content-Type", "application/json")
  211. w.Header().Set("Access-Control-Allow-Origin", "*")
  212. start := r.URL.Query().Get("start")
  213. end := r.URL.Query().Get("end")
  214. window := r.URL.Query().Get("window")
  215. fields := r.URL.Query().Get("filterFields")
  216. namespace := r.URL.Query().Get("namespace")
  217. aggregation := r.URL.Query().Get("aggregation")
  218. aggregationSubField := r.URL.Query().Get("aggregationSubfield")
  219. data, err := a.Model.ComputeCostDataRange(a.PrometheusClient, a.KubeClientSet, a.Cloud, start, end, window, namespace)
  220. if err != nil {
  221. w.Write(wrapData(nil, err))
  222. }
  223. if aggregation != "" {
  224. agg := costModel.AggregateCostModel(data, aggregation, aggregationSubField)
  225. w.Write(wrapData(agg, nil))
  226. } else {
  227. if fields != "" {
  228. filteredData := filterFields(fields, data)
  229. w.Write(wrapData(filteredData, err))
  230. } else {
  231. w.Write(wrapData(data, err))
  232. }
  233. }
  234. }
  235. // CostDataModelRangeLarge is experimental multi-cluster and long-term data storage in SQL support.
  236. func (a *Accesses) CostDataModelRangeLarge(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  237. w.Header().Set("Content-Type", "application/json")
  238. w.Header().Set("Access-Control-Allow-Origin", "*")
  239. startString := r.URL.Query().Get("start")
  240. endString := r.URL.Query().Get("end")
  241. windowString := r.URL.Query().Get("window")
  242. layout := "2006-01-02T15:04:05.000Z"
  243. var start time.Time
  244. var end time.Time
  245. var err error
  246. if windowString == "" {
  247. windowString = "1h"
  248. }
  249. if startString != "" {
  250. start, err = time.Parse(layout, startString)
  251. if err != nil {
  252. klog.V(1).Infof("Error parsing time " + startString + ". Error: " + err.Error())
  253. w.Write(wrapData(nil, err))
  254. }
  255. } else {
  256. window, err := time.ParseDuration(windowString)
  257. if err != nil {
  258. w.Write(wrapData(nil, fmt.Errorf("Invalid duration '%s'", windowString)))
  259. }
  260. start = time.Now().Add(-2 * window)
  261. }
  262. if endString != "" {
  263. end, err = time.Parse(layout, endString)
  264. if err != nil {
  265. klog.V(1).Infof("Error parsing time " + endString + ". Error: " + err.Error())
  266. w.Write(wrapData(nil, err))
  267. }
  268. } else {
  269. end = time.Now()
  270. }
  271. remoteLayout := "2006-01-02T15:04:05Z"
  272. remoteStartStr := start.Format(remoteLayout)
  273. remoteEndStr := end.Format(remoteLayout)
  274. klog.V(1).Infof("Using remote database for query from %s to %s with window %s", startString, endString, windowString)
  275. data, err := costModel.CostDataRangeFromSQL("", "", windowString, remoteStartStr, remoteEndStr)
  276. w.Write(wrapData(data, err))
  277. }
  278. func (a *Accesses) OutofClusterCosts(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  279. w.Header().Set("Content-Type", "application/json")
  280. w.Header().Set("Access-Control-Allow-Origin", "*")
  281. start := r.URL.Query().Get("start")
  282. end := r.URL.Query().Get("end")
  283. aggregator := r.URL.Query().Get("aggregator")
  284. data, err := a.Cloud.ExternalAllocations(start, end, aggregator)
  285. w.Write(wrapData(data, err))
  286. }
  287. func (p *Accesses) GetAllNodePricing(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  288. w.Header().Set("Content-Type", "application/json")
  289. w.Header().Set("Access-Control-Allow-Origin", "*")
  290. data, err := p.Cloud.AllNodePricing()
  291. w.Write(wrapData(data, err))
  292. }
  293. func (p *Accesses) GetConfigs(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  294. w.Header().Set("Content-Type", "application/json")
  295. w.Header().Set("Access-Control-Allow-Origin", "*")
  296. data, err := p.Cloud.GetConfig()
  297. w.Write(wrapData(data, err))
  298. }
  299. func (p *Accesses) UpdateSpotInfoConfigs(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  300. w.Header().Set("Content-Type", "application/json")
  301. w.Header().Set("Access-Control-Allow-Origin", "*")
  302. data, err := p.Cloud.UpdateConfig(r.Body, costAnalyzerCloud.SpotInfoUpdateType)
  303. if err != nil {
  304. w.Write(wrapData(data, err))
  305. return
  306. }
  307. w.Write(wrapData(data, err))
  308. err = p.Cloud.DownloadPricingData()
  309. if err != nil {
  310. klog.V(1).Infof("Error redownloading data on config update: %s", err.Error())
  311. }
  312. return
  313. }
  314. func (p *Accesses) UpdateAthenaInfoConfigs(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  315. w.Header().Set("Content-Type", "application/json")
  316. w.Header().Set("Access-Control-Allow-Origin", "*")
  317. data, err := p.Cloud.UpdateConfig(r.Body, costAnalyzerCloud.AthenaInfoUpdateType)
  318. if err != nil {
  319. w.Write(wrapData(data, err))
  320. return
  321. }
  322. w.Write(wrapData(data, err))
  323. return
  324. }
  325. func (p *Accesses) UpdateBigQueryInfoConfigs(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  326. w.Header().Set("Content-Type", "application/json")
  327. w.Header().Set("Access-Control-Allow-Origin", "*")
  328. data, err := p.Cloud.UpdateConfig(r.Body, costAnalyzerCloud.BigqueryUpdateType)
  329. if err != nil {
  330. w.Write(wrapData(data, err))
  331. return
  332. }
  333. w.Write(wrapData(data, err))
  334. return
  335. }
  336. func (p *Accesses) UpdateConfigByKey(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  337. w.Header().Set("Content-Type", "application/json")
  338. w.Header().Set("Access-Control-Allow-Origin", "*")
  339. data, err := p.Cloud.UpdateConfig(r.Body, "")
  340. if err != nil {
  341. w.Write(wrapData(data, err))
  342. return
  343. }
  344. w.Write(wrapData(data, err))
  345. return
  346. }
  347. func (p *Accesses) ManagementPlatform(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  348. w.Header().Set("Content-Type", "application/json")
  349. w.Header().Set("Access-Control-Allow-Origin", "*")
  350. data, err := p.Cloud.GetManagementPlatform()
  351. if err != nil {
  352. w.Write(wrapData(data, err))
  353. return
  354. }
  355. w.Write(wrapData(data, err))
  356. return
  357. }
  358. func (p *Accesses) ClusterInfo(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  359. w.Header().Set("Content-Type", "application/json")
  360. w.Header().Set("Access-Control-Allow-Origin", "*")
  361. data, err := p.Cloud.ClusterInfo()
  362. w.Write(wrapData(data, err))
  363. }
  364. func Healthz(w http.ResponseWriter, _ *http.Request, _ httprouter.Params) {
  365. w.WriteHeader(200)
  366. w.Header().Set("Content-Length", "0")
  367. w.Header().Set("Content-Type", "text/plain")
  368. }
  369. func (p *Accesses) GetPrometheusMetadata(w http.ResponseWriter, _ *http.Request, _ httprouter.Params) {
  370. w.Header().Set("Content-Type", "application/json")
  371. w.Header().Set("Access-Control-Allow-Origin", "*")
  372. w.Write(wrapData(costModel.ValidatePrometheus(p.PrometheusClient)))
  373. }
  374. func (p *Accesses) ContainerUptimes(w http.ResponseWriter, _ *http.Request, _ httprouter.Params) {
  375. w.Header().Set("Content-Type", "application/json")
  376. w.Header().Set("Access-Control-Allow-Origin", "*")
  377. res, err := costModel.ComputeUptimes(p.PrometheusClient)
  378. w.Write(wrapData(res, err))
  379. }
  380. func (a *Accesses) recordPrices() {
  381. go func() {
  382. containerSeen := make(map[string]bool)
  383. nodeSeen := make(map[string]bool)
  384. pvSeen := make(map[string]bool)
  385. getKeyFromLabelStrings := func(labels ...string) string {
  386. return strings.Join(labels, ",")
  387. }
  388. getLabelStringsFromKey := func(key string) []string {
  389. return strings.Split(key, ",")
  390. }
  391. for {
  392. klog.V(4).Info("Recording prices...")
  393. podlist := a.Model.Controller.GetAll()
  394. podStatus := make(map[string]v1.PodPhase)
  395. for _, pod := range podlist {
  396. podStatus[pod.Name] = pod.Status.Phase
  397. }
  398. data, err := a.Model.ComputeCostData(a.PrometheusClient, a.KubeClientSet, a.Cloud, "2m", "", "")
  399. if err != nil {
  400. klog.V(1).Info("Error in price recording: " + err.Error())
  401. // zero the for loop so the time.Sleep will still work
  402. data = map[string]*costModel.CostData{}
  403. }
  404. for _, costs := range data {
  405. nodeName := costs.NodeName
  406. node := costs.NodeData
  407. if node == nil {
  408. klog.V(4).Infof("Skipping Node \"%s\" due to missing Node Data costs", nodeName)
  409. continue
  410. }
  411. cpuCost, _ := strconv.ParseFloat(node.VCPUCost, 64)
  412. cpu, _ := strconv.ParseFloat(node.VCPU, 64)
  413. ramCost, _ := strconv.ParseFloat(node.RAMCost, 64)
  414. ram, _ := strconv.ParseFloat(node.RAMBytes, 64)
  415. gpu, _ := strconv.ParseFloat(node.GPU, 64)
  416. gpuCost, _ := strconv.ParseFloat(node.GPUCost, 64)
  417. totalCost := cpu*cpuCost + ramCost*(ram/1024/1024/1024) + gpu*gpuCost
  418. if costs.PVCData != nil {
  419. for _, pvc := range costs.PVCData {
  420. if pvc.Volume != nil {
  421. pvCost, _ := strconv.ParseFloat(pvc.Volume.Cost, 64)
  422. a.PersistentVolumePriceRecorder.WithLabelValues(pvc.VolumeName, pvc.VolumeName).Set(pvCost)
  423. }
  424. }
  425. }
  426. a.CPUPriceRecorder.WithLabelValues(nodeName, nodeName).Set(cpuCost)
  427. a.RAMPriceRecorder.WithLabelValues(nodeName, nodeName).Set(ramCost)
  428. a.GPUPriceRecorder.WithLabelValues(nodeName, nodeName).Set(gpuCost)
  429. a.NodeTotalPriceRecorder.WithLabelValues(nodeName, nodeName).Set(totalCost)
  430. labelKey := getKeyFromLabelStrings(nodeName, nodeName)
  431. nodeSeen[labelKey] = true
  432. namespace := costs.Namespace
  433. podName := costs.PodName
  434. containerName := costs.Name
  435. if len(costs.RAMAllocation) > 0 {
  436. a.RAMAllocationRecorder.WithLabelValues(namespace, podName, containerName, nodeName, nodeName).Set(costs.RAMAllocation[0].Value)
  437. }
  438. if len(costs.CPUAllocation) > 0 {
  439. a.CPUAllocationRecorder.WithLabelValues(namespace, podName, containerName, nodeName, nodeName).Set(costs.CPUAllocation[0].Value)
  440. }
  441. if len(costs.GPUReq) > 0 {
  442. // allocation here is set to the request because shared GPU usage not yet supported.
  443. a.GPUAllocationRecorder.WithLabelValues(namespace, podName, containerName, nodeName, nodeName).Set(costs.GPUReq[0].Value)
  444. }
  445. labelKey = getKeyFromLabelStrings(namespace, podName, containerName, nodeName, nodeName)
  446. if podStatus[podName] == v1.PodRunning { // Only report data for current pods
  447. containerSeen[labelKey] = true
  448. } else {
  449. klog.Infof("Container %s not running", labelKey)
  450. containerSeen[labelKey] = false
  451. }
  452. storageClasses, _ := a.KubeClientSet.StorageV1().StorageClasses().List(metav1.ListOptions{})
  453. storageClassMap := make(map[string]map[string]string)
  454. for _, storageClass := range storageClasses.Items {
  455. params := storageClass.Parameters
  456. storageClassMap[storageClass.ObjectMeta.Name] = params
  457. if storageClass.GetAnnotations()["storageclass.kubernetes.io/is-default-class"] == "true" || storageClass.GetAnnotations()["storageclass.beta.kubernetes.io/is-default-class"] == "true" {
  458. storageClassMap["default"] = params
  459. storageClassMap[""] = params
  460. }
  461. }
  462. pvs, _ := a.KubeClientSet.CoreV1().PersistentVolumes().List(metav1.ListOptions{})
  463. for _, pv := range pvs.Items {
  464. parameters, ok := storageClassMap[pv.Spec.StorageClassName]
  465. if !ok {
  466. klog.V(4).Infof("Unable to find parameters for storage class \"%s\". Does pv \"%s\" have a storageClassName?", pv.Spec.StorageClassName, pv.Name)
  467. }
  468. cacPv := &costAnalyzerCloud.PV{
  469. Class: pv.Spec.StorageClassName,
  470. Region: pv.Labels[v1.LabelZoneRegion],
  471. Parameters: parameters,
  472. }
  473. costModel.GetPVCost(cacPv, &pv, a.Cloud)
  474. c, _ := strconv.ParseFloat(cacPv.Cost, 64)
  475. a.PersistentVolumePriceRecorder.WithLabelValues(pv.Name, pv.Name).Set(c)
  476. labelKey := getKeyFromLabelStrings(pv.Name, pv.Name)
  477. pvSeen[labelKey] = true
  478. }
  479. containerUptime, _ := costModel.ComputeUptimes(a.PrometheusClient)
  480. for key, uptime := range containerUptime {
  481. container, _ := costModel.NewContainerMetricFromKey(key)
  482. a.ContainerUptimeRecorder.WithLabelValues(container.Namespace, container.PodName, container.ContainerName).Set(uptime)
  483. }
  484. }
  485. for labelString, seen := range nodeSeen {
  486. if !seen {
  487. labels := getLabelStringsFromKey(labelString)
  488. a.NodeTotalPriceRecorder.DeleteLabelValues(labels...)
  489. a.CPUPriceRecorder.DeleteLabelValues(labels...)
  490. a.GPUPriceRecorder.DeleteLabelValues(labels...)
  491. a.RAMPriceRecorder.DeleteLabelValues(labels...)
  492. delete(nodeSeen, labelString)
  493. }
  494. nodeSeen[labelString] = false
  495. }
  496. for labelString, seen := range containerSeen {
  497. if !seen {
  498. labels := getLabelStringsFromKey(labelString)
  499. a.RAMAllocationRecorder.DeleteLabelValues(labels...)
  500. a.CPUAllocationRecorder.DeleteLabelValues(labels...)
  501. a.GPUAllocationRecorder.DeleteLabelValues(labels...)
  502. a.ContainerUptimeRecorder.DeleteLabelValues(labels...)
  503. delete(containerSeen, labelString)
  504. }
  505. containerSeen[labelString] = false
  506. }
  507. for labelString, seen := range pvSeen {
  508. if !seen {
  509. labels := getLabelStringsFromKey(labelString)
  510. a.PersistentVolumePriceRecorder.DeleteLabelValues(labels...)
  511. delete(pvSeen, labelString)
  512. }
  513. pvSeen[labelString] = false
  514. }
  515. time.Sleep(time.Minute)
  516. }
  517. }()
  518. }
  519. func main() {
  520. klog.InitFlags(nil)
  521. flag.Set("v", "3")
  522. flag.Parse()
  523. klog.V(1).Infof("Starting cost-model (git commit \"%s\")", gitCommit)
  524. address := os.Getenv(prometheusServerEndpointEnvVar)
  525. if address == "" {
  526. klog.Fatalf("No address for prometheus set in $%s. Aborting.", prometheusServerEndpointEnvVar)
  527. }
  528. var LongTimeoutRoundTripper http.RoundTripper = &http.Transport{ // may be necessary for long prometheus queries. TODO: make this configurable
  529. Proxy: http.ProxyFromEnvironment,
  530. DialContext: (&net.Dialer{
  531. Timeout: 120 * time.Second,
  532. KeepAlive: 120 * time.Second,
  533. }).DialContext,
  534. TLSHandshakeTimeout: 10 * time.Second,
  535. }
  536. pc := prometheusClient.Config{
  537. Address: address,
  538. RoundTripper: LongTimeoutRoundTripper,
  539. }
  540. promCli, _ := prometheusClient.NewClient(pc)
  541. api := prometheusAPI.NewAPI(promCli)
  542. _, err := api.Config(context.Background())
  543. if err != nil {
  544. klog.Fatalf("No valid prometheus config file at %s. Error: %s . Troubleshooting help available at: %s", address, err.Error(), prometheusTroubleshootingEp)
  545. }
  546. klog.V(1).Info("Success: retrieved a prometheus config file from: " + address)
  547. _, err = costModel.ValidatePrometheus(promCli)
  548. if err != nil {
  549. klog.Fatalf("Failed to query prometheus at %s. Error: %s . Troubleshooting help available at: %s", address, err.Error(), prometheusTroubleshootingEp)
  550. }
  551. klog.V(1).Info("Success: retrieved the 'up' query against prometheus at: " + address)
  552. // Kubernetes API setup
  553. kc, err := rest.InClusterConfig()
  554. if err != nil {
  555. panic(err.Error())
  556. }
  557. kubeClientset, err := kubernetes.NewForConfig(kc)
  558. if err != nil {
  559. panic(err.Error())
  560. }
  561. cloudProviderKey := os.Getenv("CLOUD_PROVIDER_API_KEY")
  562. cloudProvider, err := costAnalyzerCloud.NewProvider(kubeClientset, cloudProviderKey)
  563. if err != nil {
  564. panic(err.Error())
  565. }
  566. cpuGv := prometheus.NewGaugeVec(prometheus.GaugeOpts{
  567. Name: "node_cpu_hourly_cost",
  568. Help: "node_cpu_hourly_cost hourly cost for each cpu on this node",
  569. }, []string{"instance", "node"})
  570. ramGv := prometheus.NewGaugeVec(prometheus.GaugeOpts{
  571. Name: "node_ram_hourly_cost",
  572. Help: "node_ram_hourly_cost hourly cost for each gb of ram on this node",
  573. }, []string{"instance", "node"})
  574. gpuGv := prometheus.NewGaugeVec(prometheus.GaugeOpts{
  575. Name: "node_gpu_hourly_cost",
  576. Help: "node_gpu_hourly_cost hourly cost for each gpu on this node",
  577. }, []string{"instance", "node"})
  578. totalGv := prometheus.NewGaugeVec(prometheus.GaugeOpts{
  579. Name: "node_total_hourly_cost",
  580. Help: "node_total_hourly_cost Total node cost per hour",
  581. }, []string{"instance", "node"})
  582. pvGv := prometheus.NewGaugeVec(prometheus.GaugeOpts{
  583. Name: "pv_hourly_cost",
  584. Help: "pv_hourly_cost Cost per GB per hour on a persistent disk",
  585. }, []string{"volumename", "persistentvolume"})
  586. RAMAllocation := prometheus.NewGaugeVec(prometheus.GaugeOpts{
  587. Name: "container_memory_allocation_bytes",
  588. Help: "container_memory_allocation_bytes Bytes of RAM used",
  589. }, []string{"namespace", "pod", "container", "instance", "node"})
  590. CPUAllocation := prometheus.NewGaugeVec(prometheus.GaugeOpts{
  591. Name: "container_cpu_allocation",
  592. Help: "container_cpu_allocation Percent of a single CPU used in a minute",
  593. }, []string{"namespace", "pod", "container", "instance", "node"})
  594. GPUAllocation := prometheus.NewGaugeVec(prometheus.GaugeOpts{
  595. Name: "container_gpu_allocation",
  596. Help: "container_gpu_allocation GPU used",
  597. }, []string{"namespace", "pod", "container", "instance", "node"})
  598. ContainerUptimeRecorder := prometheus.NewGaugeVec(prometheus.GaugeOpts{
  599. Name: "container_uptime_seconds",
  600. Help: "container_uptime_seconds Seconds a container has been running",
  601. }, []string{"namespace", "pod", "container"})
  602. prometheus.MustRegister(cpuGv)
  603. prometheus.MustRegister(ramGv)
  604. prometheus.MustRegister(gpuGv)
  605. prometheus.MustRegister(totalGv)
  606. prometheus.MustRegister(pvGv)
  607. prometheus.MustRegister(RAMAllocation)
  608. prometheus.MustRegister(CPUAllocation)
  609. prometheus.MustRegister(ContainerUptimeRecorder)
  610. prometheus.MustRegister(costModel.ServiceCollector{
  611. KubeClientSet: kubeClientset,
  612. })
  613. prometheus.MustRegister(costModel.DeploymentCollector{
  614. KubeClientSet: kubeClientset,
  615. })
  616. podCache := cache.NewListWatchFromClient(kubeClientset.CoreV1().RESTClient(), "pods", "", fields.Everything())
  617. a := Accesses{
  618. PrometheusClient: promCli,
  619. KubeClientSet: kubeClientset,
  620. Cloud: cloudProvider,
  621. CPUPriceRecorder: cpuGv,
  622. RAMPriceRecorder: ramGv,
  623. GPUPriceRecorder: gpuGv,
  624. NodeTotalPriceRecorder: totalGv,
  625. RAMAllocationRecorder: RAMAllocation,
  626. CPUAllocationRecorder: CPUAllocation,
  627. GPUAllocationRecorder: GPUAllocation,
  628. ContainerUptimeRecorder: ContainerUptimeRecorder,
  629. PersistentVolumePriceRecorder: pvGv,
  630. Model: costModel.NewCostModel(podCache),
  631. }
  632. remoteEnabled := os.Getenv(remoteEnabled)
  633. if remoteEnabled == "true" {
  634. info, err := cloudProvider.ClusterInfo()
  635. klog.Infof("Saving cluster with id:'%s', and name:'%s' to durable storage", info["id"], info["name"])
  636. if err != nil {
  637. klog.Infof("Error saving cluster id %s", err.Error())
  638. }
  639. _, _, err = costAnalyzerCloud.GetOrCreateClusterMeta(info["id"], info["name"])
  640. if err != nil {
  641. klog.Infof("Unable to set cluster id '%s' for cluster '%s', %s", info["id"], info["name"], err.Error())
  642. }
  643. }
  644. err = a.Cloud.DownloadPricingData()
  645. if err != nil {
  646. klog.V(1).Info("Failed to download pricing data: " + err.Error())
  647. }
  648. a.recordPrices()
  649. router := httprouter.New()
  650. router.GET("/costDataModel", a.CostDataModel)
  651. router.GET("/costDataModelRange", a.CostDataModelRange)
  652. router.GET("/costDataModelRangeLarge", a.CostDataModelRangeLarge)
  653. router.GET("/outOfClusterCosts", a.OutofClusterCosts)
  654. router.GET("/allNodePricing", a.GetAllNodePricing)
  655. router.GET("/healthz", Healthz)
  656. router.GET("/getConfigs", a.GetConfigs)
  657. router.POST("/refreshPricing", a.RefreshPricingData)
  658. router.POST("/updateSpotInfoConfigs", a.UpdateSpotInfoConfigs)
  659. router.POST("/updateAthenaInfoConfigs", a.UpdateAthenaInfoConfigs)
  660. router.POST("/updateBigQueryInfoConfigs", a.UpdateBigQueryInfoConfigs)
  661. router.POST("/updateConfigByKey", a.UpdateConfigByKey)
  662. router.GET("/clusterCostsOverTime", a.ClusterCostsOverTime)
  663. router.GET("/clusterCosts", a.ClusterCosts)
  664. router.GET("/validatePrometheus", a.GetPrometheusMetadata)
  665. router.GET("/managementPlatform", a.ManagementPlatform)
  666. router.GET("/clusterInfo", a.ClusterInfo)
  667. router.GET("/containerUptimes", a.ContainerUptimes)
  668. router.GET("/aggregatedCostModel", a.AggregateCostModel)
  669. rootMux := http.NewServeMux()
  670. rootMux.Handle("/", router)
  671. rootMux.Handle("/metrics", promhttp.Handler())
  672. klog.Fatal(http.ListenAndServe(":9003", rootMux))
  673. }