main.go 26 KB

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