main.go 30 KB

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