main.go 29 KB

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