router.go 43 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250
  1. package costmodel
  2. import (
  3. "context"
  4. "encoding/json"
  5. "flag"
  6. "fmt"
  7. "math"
  8. "net"
  9. "net/http"
  10. "reflect"
  11. "strconv"
  12. "strings"
  13. "time"
  14. "k8s.io/klog"
  15. "github.com/julienschmidt/httprouter"
  16. sentry "github.com/getsentry/sentry-go"
  17. costAnalyzerCloud "github.com/kubecost/cost-model/pkg/cloud"
  18. "github.com/kubecost/cost-model/pkg/clustercache"
  19. cm "github.com/kubecost/cost-model/pkg/clustermanager"
  20. "github.com/kubecost/cost-model/pkg/env"
  21. "github.com/kubecost/cost-model/pkg/errors"
  22. "github.com/kubecost/cost-model/pkg/prom"
  23. "github.com/kubecost/cost-model/pkg/thanos"
  24. prometheusClient "github.com/prometheus/client_golang/api"
  25. prometheusAPI "github.com/prometheus/client_golang/api/prometheus/v1"
  26. v1 "k8s.io/api/core/v1"
  27. metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
  28. "github.com/patrickmn/go-cache"
  29. "github.com/prometheus/client_golang/prometheus"
  30. "k8s.io/client-go/kubernetes"
  31. "k8s.io/client-go/rest"
  32. )
  33. const (
  34. prometheusTroubleshootingEp = "http://docs.kubecost.com/custom-prom#troubleshoot"
  35. RFC3339Milli = "2006-01-02T15:04:05.000Z"
  36. )
  37. var (
  38. // gitCommit is set by the build system
  39. gitCommit string
  40. logCollectionEnabled bool = env.IsLogCollectionEnabled()
  41. productAnalyticsEnabled bool = env.IsProductAnalyticsEnabled()
  42. errorReportingEnabled bool = env.IsErrorReportingEnabled()
  43. valuesReportingEnabled bool = env.IsValuesReportingEnabled()
  44. clusterProfile string = env.GetClusterProfile()
  45. multiclusterDBBasicAuthUsername string = env.GetMultiClusterBasicAuthUsername()
  46. multiclusterDBBasicAuthPW string = env.GetMultiClusterBasicAuthPassword()
  47. )
  48. var Router = httprouter.New()
  49. var A Accesses
  50. type Accesses struct {
  51. PrometheusClient prometheusClient.Client
  52. ThanosClient prometheusClient.Client
  53. KubeClientSet kubernetes.Interface
  54. ClusterManager *cm.ClusterManager
  55. Cloud costAnalyzerCloud.Provider
  56. CPUPriceRecorder *prometheus.GaugeVec
  57. RAMPriceRecorder *prometheus.GaugeVec
  58. PersistentVolumePriceRecorder *prometheus.GaugeVec
  59. GPUPriceRecorder *prometheus.GaugeVec
  60. NodeTotalPriceRecorder *prometheus.GaugeVec
  61. RAMAllocationRecorder *prometheus.GaugeVec
  62. CPUAllocationRecorder *prometheus.GaugeVec
  63. GPUAllocationRecorder *prometheus.GaugeVec
  64. PVAllocationRecorder *prometheus.GaugeVec
  65. NetworkZoneEgressRecorder prometheus.Gauge
  66. NetworkRegionEgressRecorder prometheus.Gauge
  67. NetworkInternetEgressRecorder prometheus.Gauge
  68. ServiceSelectorRecorder *prometheus.GaugeVec
  69. DeploymentSelectorRecorder *prometheus.GaugeVec
  70. Model *CostModel
  71. OutOfClusterCache *cache.Cache
  72. }
  73. type DataEnvelope struct {
  74. Code int `json:"code"`
  75. Status string `json:"status"`
  76. Data interface{} `json:"data"`
  77. Message string `json:"message,omitempty"`
  78. }
  79. // FilterFunc is a filter that returns true iff the given CostData should be filtered out, and the environment that was used as the filter criteria, if it was an aggregate
  80. type FilterFunc func(*CostData) (bool, string)
  81. // FilterCostData allows through only CostData that matches all the given filter functions
  82. func FilterCostData(data map[string]*CostData, retains []FilterFunc, filters []FilterFunc) (map[string]*CostData, int, map[string]int) {
  83. result := make(map[string]*CostData)
  84. filteredEnvironments := make(map[string]int)
  85. filteredContainers := 0
  86. DataLoop:
  87. for key, datum := range data {
  88. for _, rf := range retains {
  89. if ok, _ := rf(datum); ok {
  90. result[key] = datum
  91. // if any retain function passes, the data is retained and move on
  92. continue DataLoop
  93. }
  94. }
  95. for _, ff := range filters {
  96. if ok, environment := ff(datum); !ok {
  97. if environment != "" {
  98. filteredEnvironments[environment]++
  99. }
  100. filteredContainers++
  101. // if any filter function check fails, move on to the next datum
  102. continue DataLoop
  103. }
  104. }
  105. result[key] = datum
  106. }
  107. return result, filteredContainers, filteredEnvironments
  108. }
  109. func filterFields(fields string, data map[string]*CostData) map[string]CostData {
  110. fs := strings.Split(fields, ",")
  111. fmap := make(map[string]bool)
  112. for _, f := range fs {
  113. fieldNameLower := strings.ToLower(f) // convert to go struct name by uppercasing first letter
  114. klog.V(1).Infof("to delete: %s", fieldNameLower)
  115. fmap[fieldNameLower] = true
  116. }
  117. filteredData := make(map[string]CostData)
  118. for cname, costdata := range data {
  119. s := reflect.TypeOf(*costdata)
  120. val := reflect.ValueOf(*costdata)
  121. costdata2 := CostData{}
  122. cd2 := reflect.New(reflect.Indirect(reflect.ValueOf(costdata2)).Type()).Elem()
  123. n := s.NumField()
  124. for i := 0; i < n; i++ {
  125. field := s.Field(i)
  126. value := val.Field(i)
  127. value2 := cd2.Field(i)
  128. if _, ok := fmap[strings.ToLower(field.Name)]; !ok {
  129. value2.Set(reflect.Value(value))
  130. }
  131. }
  132. filteredData[cname] = cd2.Interface().(CostData)
  133. }
  134. return filteredData
  135. }
  136. func normalizeTimeParam(param string) (string, error) {
  137. if param == "" {
  138. return "", fmt.Errorf("invalid time param")
  139. }
  140. // convert days to hours
  141. if param[len(param)-1:] == "d" {
  142. count := param[:len(param)-1]
  143. val, err := strconv.ParseInt(count, 10, 64)
  144. if err != nil {
  145. return "", err
  146. }
  147. val = val * 24
  148. param = fmt.Sprintf("%dh", val)
  149. }
  150. return param, nil
  151. }
  152. // writeReportingFlags writes the reporting flags to the cluster info map
  153. func writeReportingFlags(clusterInfo map[string]string) {
  154. clusterInfo["logCollection"] = fmt.Sprintf("%t", logCollectionEnabled)
  155. clusterInfo["productAnalytics"] = fmt.Sprintf("%t", productAnalyticsEnabled)
  156. clusterInfo["errorReporting"] = fmt.Sprintf("%t", errorReportingEnabled)
  157. clusterInfo["valuesReporting"] = fmt.Sprintf("%t", valuesReportingEnabled)
  158. }
  159. // parsePercentString takes a string of expected format "N%" and returns a floating point 0.0N.
  160. // If the "%" symbol is missing, it just returns 0.0N. Empty string is interpreted as "0%" and
  161. // return 0.0.
  162. func ParsePercentString(percentStr string) (float64, error) {
  163. if len(percentStr) == 0 {
  164. return 0.0, nil
  165. }
  166. if percentStr[len(percentStr)-1:] == "%" {
  167. percentStr = percentStr[:len(percentStr)-1]
  168. }
  169. discount, err := strconv.ParseFloat(percentStr, 64)
  170. if err != nil {
  171. return 0.0, err
  172. }
  173. discount *= 0.01
  174. return discount, nil
  175. }
  176. // parseDuration converts a Prometheus-style duration string into a Duration
  177. func ParseDuration(duration string) (*time.Duration, error) {
  178. unitStr := duration[len(duration)-1:]
  179. var unit time.Duration
  180. switch unitStr {
  181. case "s":
  182. unit = time.Second
  183. case "m":
  184. unit = time.Minute
  185. case "h":
  186. unit = time.Hour
  187. case "d":
  188. unit = 24.0 * time.Hour
  189. default:
  190. return nil, fmt.Errorf("error parsing duration: %s did not match expected format [0-9+](s|m|d|h)", duration)
  191. }
  192. amountStr := duration[:len(duration)-1]
  193. amount, err := strconv.ParseInt(amountStr, 10, 64)
  194. if err != nil {
  195. return nil, fmt.Errorf("error parsing duration: %s did not match expected format [0-9+](s|m|d|h)", duration)
  196. }
  197. dur := time.Duration(amount) * unit
  198. return &dur, nil
  199. }
  200. // ParseTimeRange returns a start and end time, respectively, which are converted from
  201. // a duration and offset, defined as strings with Prometheus-style syntax.
  202. func ParseTimeRange(duration, offset string) (*time.Time, *time.Time, error) {
  203. // endTime defaults to the current time, unless an offset is explicity declared,
  204. // in which case it shifts endTime back by given duration
  205. endTime := time.Now()
  206. if offset != "" {
  207. o, err := ParseDuration(offset)
  208. if err != nil {
  209. return nil, nil, fmt.Errorf("error parsing offset (%s): %s", offset, err)
  210. }
  211. endTime = endTime.Add(-1 * *o)
  212. }
  213. // if duration is defined in terms of days, convert to hours
  214. // e.g. convert "2d" to "48h"
  215. durationNorm, err := normalizeTimeParam(duration)
  216. if err != nil {
  217. return nil, nil, fmt.Errorf("error parsing duration (%s): %s", duration, err)
  218. }
  219. // convert time duration into start and end times, formatted
  220. // as ISO datetime strings
  221. dur, err := time.ParseDuration(durationNorm)
  222. if err != nil {
  223. return nil, nil, fmt.Errorf("errorf parsing duration (%s): %s", durationNorm, err)
  224. }
  225. startTime := endTime.Add(-1 * dur)
  226. return &startTime, &endTime, nil
  227. }
  228. func WrapDataWithMessage(data interface{}, err error, message string) []byte {
  229. var resp []byte
  230. if err != nil {
  231. klog.V(1).Infof("Error returned to client: %s", err.Error())
  232. resp, _ = json.Marshal(&DataEnvelope{
  233. Code: http.StatusInternalServerError,
  234. Status: "error",
  235. Message: err.Error(),
  236. Data: data,
  237. })
  238. } else {
  239. resp, _ = json.Marshal(&DataEnvelope{
  240. Code: http.StatusOK,
  241. Status: "success",
  242. Data: data,
  243. Message: message,
  244. })
  245. }
  246. return resp
  247. }
  248. func WrapData(data interface{}, err error) []byte {
  249. var resp []byte
  250. if err != nil {
  251. klog.V(1).Infof("Error returned to client: %s", err.Error())
  252. resp, _ = json.Marshal(&DataEnvelope{
  253. Code: http.StatusInternalServerError,
  254. Status: "error",
  255. Message: err.Error(),
  256. Data: data,
  257. })
  258. } else {
  259. resp, _ = json.Marshal(&DataEnvelope{
  260. Code: http.StatusOK,
  261. Status: "success",
  262. Data: data,
  263. })
  264. }
  265. return resp
  266. }
  267. // 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.
  268. func (a *Accesses) RefreshPricingData(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  269. w.Header().Set("Content-Type", "application/json")
  270. w.Header().Set("Access-Control-Allow-Origin", "*")
  271. err := a.Cloud.DownloadPricingData()
  272. w.Write(WrapData(nil, err))
  273. }
  274. func (a *Accesses) CostDataModel(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. window := r.URL.Query().Get("timeWindow")
  278. offset := r.URL.Query().Get("offset")
  279. fields := r.URL.Query().Get("filterFields")
  280. namespace := r.URL.Query().Get("namespace")
  281. if offset != "" {
  282. offset = "offset " + offset
  283. }
  284. data, err := a.Model.ComputeCostData(a.PrometheusClient, a.KubeClientSet, a.Cloud, window, offset, namespace)
  285. if fields != "" {
  286. filteredData := filterFields(fields, data)
  287. w.Write(WrapData(filteredData, err))
  288. } else {
  289. w.Write(WrapData(data, err))
  290. }
  291. }
  292. func (a *Accesses) ClusterCosts(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  293. w.Header().Set("Content-Type", "application/json")
  294. w.Header().Set("Access-Control-Allow-Origin", "*")
  295. window := r.URL.Query().Get("window")
  296. offset := r.URL.Query().Get("offset")
  297. data, err := ComputeClusterCosts(a.PrometheusClient, a.Cloud, window, offset, true)
  298. w.Write(WrapData(data, err))
  299. }
  300. func (a *Accesses) ClusterCostsOverTime(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  301. w.Header().Set("Content-Type", "application/json")
  302. w.Header().Set("Access-Control-Allow-Origin", "*")
  303. start := r.URL.Query().Get("start")
  304. end := r.URL.Query().Get("end")
  305. window := r.URL.Query().Get("window")
  306. offset := r.URL.Query().Get("offset")
  307. data, err := ClusterCostsOverTime(a.PrometheusClient, a.Cloud, start, end, window, offset)
  308. w.Write(WrapData(data, err))
  309. }
  310. func (a *Accesses) CostDataModelRange(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  311. w.Header().Set("Content-Type", "application/json")
  312. w.Header().Set("Access-Control-Allow-Origin", "*")
  313. start := r.URL.Query().Get("start")
  314. end := r.URL.Query().Get("end")
  315. window := r.URL.Query().Get("window")
  316. fields := r.URL.Query().Get("filterFields")
  317. namespace := r.URL.Query().Get("namespace")
  318. cluster := r.URL.Query().Get("cluster")
  319. remote := r.URL.Query().Get("remote")
  320. remoteEnabled := env.IsRemoteEnabled() && remote != "false"
  321. // Use Thanos Client if it exists (enabled) and remote flag set
  322. var pClient prometheusClient.Client
  323. if remote != "false" && a.ThanosClient != nil {
  324. pClient = a.ThanosClient
  325. } else {
  326. pClient = a.PrometheusClient
  327. }
  328. resolutionHours := 1.0
  329. data, err := a.Model.ComputeCostDataRange(pClient, a.KubeClientSet, a.Cloud, start, end, window, resolutionHours, namespace, cluster, remoteEnabled)
  330. if err != nil {
  331. w.Write(WrapData(nil, err))
  332. }
  333. if fields != "" {
  334. filteredData := filterFields(fields, data)
  335. w.Write(WrapData(filteredData, err))
  336. } else {
  337. w.Write(WrapData(data, err))
  338. }
  339. }
  340. // CostDataModelRangeLarge is experimental multi-cluster and long-term data storage in SQL support.
  341. func (a *Accesses) CostDataModelRangeLarge(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  342. w.Header().Set("Content-Type", "application/json")
  343. w.Header().Set("Access-Control-Allow-Origin", "*")
  344. startString := r.URL.Query().Get("start")
  345. endString := r.URL.Query().Get("end")
  346. windowString := r.URL.Query().Get("window")
  347. var start time.Time
  348. var end time.Time
  349. var err error
  350. if windowString == "" {
  351. windowString = "1h"
  352. }
  353. if startString != "" {
  354. start, err = time.Parse(RFC3339Milli, startString)
  355. if err != nil {
  356. klog.V(1).Infof("Error parsing time " + startString + ". Error: " + err.Error())
  357. w.Write(WrapData(nil, err))
  358. }
  359. } else {
  360. window, err := time.ParseDuration(windowString)
  361. if err != nil {
  362. w.Write(WrapData(nil, fmt.Errorf("Invalid duration '%s'", windowString)))
  363. }
  364. start = time.Now().Add(-2 * window)
  365. }
  366. if endString != "" {
  367. end, err = time.Parse(RFC3339Milli, endString)
  368. if err != nil {
  369. klog.V(1).Infof("Error parsing time " + endString + ". Error: " + err.Error())
  370. w.Write(WrapData(nil, err))
  371. }
  372. } else {
  373. end = time.Now()
  374. }
  375. remoteLayout := "2006-01-02T15:04:05Z"
  376. remoteStartStr := start.Format(remoteLayout)
  377. remoteEndStr := end.Format(remoteLayout)
  378. klog.V(1).Infof("Using remote database for query from %s to %s with window %s", startString, endString, windowString)
  379. data, err := CostDataRangeFromSQL("", "", windowString, remoteStartStr, remoteEndStr)
  380. w.Write(WrapData(data, err))
  381. }
  382. func parseAggregations(customAggregation, aggregator, filterType string) (string, []string, string) {
  383. var key string
  384. var filter string
  385. var val []string
  386. if customAggregation != "" {
  387. key = customAggregation
  388. filter = filterType
  389. val = strings.Split(customAggregation, ",")
  390. } else {
  391. aggregations := strings.Split(aggregator, ",")
  392. for i, agg := range aggregations {
  393. aggregations[i] = "kubernetes_" + agg
  394. }
  395. key = strings.Join(aggregations, ",")
  396. filter = "kubernetes_" + filterType
  397. val = aggregations
  398. }
  399. return key, val, filter
  400. }
  401. func (a *Accesses) OutofClusterCosts(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  402. w.Header().Set("Content-Type", "application/json")
  403. w.Header().Set("Access-Control-Allow-Origin", "*")
  404. start := r.URL.Query().Get("start")
  405. end := r.URL.Query().Get("end")
  406. aggregator := r.URL.Query().Get("aggregator")
  407. customAggregation := r.URL.Query().Get("customAggregation")
  408. filterType := r.URL.Query().Get("filterType")
  409. filterValue := r.URL.Query().Get("filterValue")
  410. var data []*costAnalyzerCloud.OutOfClusterAllocation
  411. var err error
  412. _, aggregations, filter := parseAggregations(customAggregation, aggregator, filterType)
  413. data, err = a.Cloud.ExternalAllocations(start, end, aggregations, filter, filterValue, false)
  414. w.Write(WrapData(data, err))
  415. }
  416. func (a *Accesses) OutOfClusterCostsWithCache(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  417. w.Header().Set("Content-Type", "application/json")
  418. w.Header().Set("Access-Control-Allow-Origin", "*")
  419. // start date for which to query costs, inclusive; format YYYY-MM-DD
  420. start := r.URL.Query().Get("start")
  421. // end date for which to query costs, inclusive; format YYYY-MM-DD
  422. end := r.URL.Query().Get("end")
  423. // aggregator sets the field by which to aggregate; default, prepended by "kubernetes_"
  424. kubernetesAggregation := r.URL.Query().Get("aggregator")
  425. // customAggregation allows full customization of aggregator w/o prepending
  426. customAggregation := r.URL.Query().Get("customAggregation")
  427. // disableCache, if set to "true", tells this function to recompute and
  428. // cache the requested data
  429. disableCache := r.URL.Query().Get("disableCache") == "true"
  430. // clearCache, if set to "true", tells this function to flush the cache,
  431. // then recompute and cache the requested data
  432. clearCache := r.URL.Query().Get("clearCache") == "true"
  433. filterType := r.URL.Query().Get("filterType")
  434. filterValue := r.URL.Query().Get("filterValue")
  435. aggregationkey, aggregation, filter := parseAggregations(customAggregation, kubernetesAggregation, filterType)
  436. // clear cache prior to checking the cache so that a clearCache=true
  437. // request always returns a freshly computed value
  438. if clearCache {
  439. a.OutOfClusterCache.Flush()
  440. }
  441. // attempt to retrieve cost data from cache
  442. key := fmt.Sprintf(`%s:%s:%s:%s:%s`, start, end, aggregationkey, filter, filterValue)
  443. if value, found := a.OutOfClusterCache.Get(key); found && !disableCache {
  444. if data, ok := value.([]*costAnalyzerCloud.OutOfClusterAllocation); ok {
  445. w.Write(WrapDataWithMessage(data, nil, fmt.Sprintf("out of cluster cache hit: %s", key)))
  446. return
  447. }
  448. klog.Errorf("caching error: failed to type cast data: %s", key)
  449. }
  450. data, err := a.Cloud.ExternalAllocations(start, end, aggregation, filter, filterValue, false)
  451. if err == nil {
  452. a.OutOfClusterCache.Set(key, data, cache.DefaultExpiration)
  453. }
  454. w.Write(WrapDataWithMessage(data, err, fmt.Sprintf("out of cluser cache miss: %s", key)))
  455. }
  456. func (p *Accesses) GetAllNodePricing(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  457. w.Header().Set("Content-Type", "application/json")
  458. w.Header().Set("Access-Control-Allow-Origin", "*")
  459. data, err := p.Cloud.AllNodePricing()
  460. w.Write(WrapData(data, err))
  461. }
  462. func (p *Accesses) GetConfigs(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  463. w.Header().Set("Content-Type", "application/json")
  464. w.Header().Set("Access-Control-Allow-Origin", "*")
  465. data, err := p.Cloud.GetConfig()
  466. w.Write(WrapData(data, err))
  467. }
  468. func (p *Accesses) UpdateSpotInfoConfigs(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  469. w.Header().Set("Content-Type", "application/json")
  470. w.Header().Set("Access-Control-Allow-Origin", "*")
  471. data, err := p.Cloud.UpdateConfig(r.Body, costAnalyzerCloud.SpotInfoUpdateType)
  472. if err != nil {
  473. w.Write(WrapData(data, err))
  474. return
  475. }
  476. w.Write(WrapData(data, err))
  477. err = p.Cloud.DownloadPricingData()
  478. if err != nil {
  479. klog.V(1).Infof("Error redownloading data on config update: %s", err.Error())
  480. }
  481. return
  482. }
  483. func (p *Accesses) UpdateAthenaInfoConfigs(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  484. w.Header().Set("Content-Type", "application/json")
  485. w.Header().Set("Access-Control-Allow-Origin", "*")
  486. data, err := p.Cloud.UpdateConfig(r.Body, costAnalyzerCloud.AthenaInfoUpdateType)
  487. if err != nil {
  488. w.Write(WrapData(data, err))
  489. return
  490. }
  491. w.Write(WrapData(data, err))
  492. return
  493. }
  494. func (p *Accesses) UpdateBigQueryInfoConfigs(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  495. w.Header().Set("Content-Type", "application/json")
  496. w.Header().Set("Access-Control-Allow-Origin", "*")
  497. data, err := p.Cloud.UpdateConfig(r.Body, costAnalyzerCloud.BigqueryUpdateType)
  498. if err != nil {
  499. w.Write(WrapData(data, err))
  500. return
  501. }
  502. w.Write(WrapData(data, err))
  503. return
  504. }
  505. func (p *Accesses) UpdateConfigByKey(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  506. w.Header().Set("Content-Type", "application/json")
  507. w.Header().Set("Access-Control-Allow-Origin", "*")
  508. data, err := p.Cloud.UpdateConfig(r.Body, "")
  509. if err != nil {
  510. w.Write(WrapData(data, err))
  511. return
  512. }
  513. w.Write(WrapData(data, err))
  514. return
  515. }
  516. func (p *Accesses) ManagementPlatform(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  517. w.Header().Set("Content-Type", "application/json")
  518. w.Header().Set("Access-Control-Allow-Origin", "*")
  519. data, err := p.Cloud.GetManagementPlatform()
  520. if err != nil {
  521. w.Write(WrapData(data, err))
  522. return
  523. }
  524. w.Write(WrapData(data, err))
  525. return
  526. }
  527. func (p *Accesses) ClusterInfo(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
  528. w.Header().Set("Content-Type", "application/json")
  529. w.Header().Set("Access-Control-Allow-Origin", "*")
  530. data, err := p.Cloud.ClusterInfo()
  531. kc, ok := p.KubeClientSet.(*kubernetes.Clientset)
  532. if ok && data != nil {
  533. v, err := kc.ServerVersion()
  534. if err != nil {
  535. klog.Infof("Could not get k8s version info: %s", err.Error())
  536. } else if v != nil {
  537. data["version"] = v.Major + "." + v.Minor
  538. }
  539. } else {
  540. klog.Infof("Could not get k8s version info: %s", err.Error())
  541. }
  542. // Ensure we create the info object if it doesn't exist
  543. if data == nil {
  544. data = make(map[string]string)
  545. }
  546. data["clusterProfile"] = clusterProfile
  547. // Include Product Reporting Flags with Cluster Info
  548. writeReportingFlags(data)
  549. // Include Thanos Offset Duration if Applicable
  550. if thanos.IsEnabled() {
  551. data["thanosOffset"] = thanos.Offset()
  552. }
  553. w.Write(WrapData(data, err))
  554. }
  555. func (p *Accesses) GetServiceAccountStatus(w http.ResponseWriter, _ *http.Request, _ httprouter.Params) {
  556. w.Header().Set("Content-Type", "application/json")
  557. w.Header().Set("Access-Control-Allow-Origin", "*")
  558. w.Write(WrapData(A.Cloud.ServiceAccountStatus(), nil))
  559. }
  560. func (p *Accesses) GetPrometheusMetadata(w http.ResponseWriter, _ *http.Request, _ httprouter.Params) {
  561. w.Header().Set("Content-Type", "application/json")
  562. w.Header().Set("Access-Control-Allow-Origin", "*")
  563. w.Write(WrapData(ValidatePrometheus(p.PrometheusClient, false)))
  564. }
  565. func (a *Accesses) recordPrices() {
  566. go func() {
  567. defer errors.HandlePanic()
  568. containerSeen := make(map[string]bool)
  569. nodeSeen := make(map[string]bool)
  570. pvSeen := make(map[string]bool)
  571. pvcSeen := make(map[string]bool)
  572. getKeyFromLabelStrings := func(labels ...string) string {
  573. return strings.Join(labels, ",")
  574. }
  575. getLabelStringsFromKey := func(key string) []string {
  576. return strings.Split(key, ",")
  577. }
  578. var defaultRegion string = ""
  579. nodeList := a.Model.Cache.GetAllNodes()
  580. if len(nodeList) > 0 {
  581. defaultRegion = nodeList[0].Labels[v1.LabelZoneRegion]
  582. }
  583. for {
  584. klog.V(4).Info("Recording prices...")
  585. podlist := a.Model.Cache.GetAllPods()
  586. podStatus := make(map[string]v1.PodPhase)
  587. for _, pod := range podlist {
  588. podStatus[pod.Name] = pod.Status.Phase
  589. }
  590. cfg, _ := a.Cloud.GetConfig()
  591. // Record network pricing at global scope
  592. networkCosts, err := a.Cloud.NetworkPricing()
  593. if err != nil {
  594. klog.V(4).Infof("Failed to retrieve network costs: %s", err.Error())
  595. } else {
  596. a.NetworkZoneEgressRecorder.Set(networkCosts.ZoneNetworkEgressCost)
  597. a.NetworkRegionEgressRecorder.Set(networkCosts.RegionNetworkEgressCost)
  598. a.NetworkInternetEgressRecorder.Set(networkCosts.InternetNetworkEgressCost)
  599. }
  600. data, err := a.Model.ComputeCostData(a.PrometheusClient, a.KubeClientSet, a.Cloud, "2m", "", "")
  601. if err != nil {
  602. klog.V(1).Info("Error in price recording: " + err.Error())
  603. // zero the for loop so the time.Sleep will still work
  604. data = map[string]*CostData{}
  605. }
  606. nodes, err := a.Model.GetNodeCost(a.Cloud)
  607. for nodeName, node := range nodes {
  608. // Emit costs, guarding against NaN inputs for custom pricing.
  609. cpuCost, _ := strconv.ParseFloat(node.VCPUCost, 64)
  610. if math.IsNaN(cpuCost) || math.IsInf(cpuCost, 0) {
  611. cpuCost, _ = strconv.ParseFloat(cfg.CPU, 64)
  612. if math.IsNaN(cpuCost) || math.IsInf(cpuCost, 0) {
  613. cpuCost = 0
  614. }
  615. }
  616. cpu, _ := strconv.ParseFloat(node.VCPU, 64)
  617. if math.IsNaN(cpu) || math.IsInf(cpu, 0) {
  618. cpu = 1 // Assume 1 CPU
  619. }
  620. ramCost, _ := strconv.ParseFloat(node.RAMCost, 64)
  621. if math.IsNaN(ramCost) || math.IsInf(ramCost, 0) {
  622. ramCost, _ = strconv.ParseFloat(cfg.RAM, 64)
  623. if math.IsNaN(ramCost) || math.IsInf(ramCost, 0) {
  624. ramCost = 0
  625. }
  626. }
  627. ram, _ := strconv.ParseFloat(node.RAMBytes, 64)
  628. if math.IsNaN(ram) || math.IsInf(ram, 0) {
  629. ram = 0
  630. }
  631. gpu, _ := strconv.ParseFloat(node.GPU, 64)
  632. if math.IsNaN(gpu) || math.IsInf(gpu, 0) {
  633. gpu = 0
  634. }
  635. gpuCost, _ := strconv.ParseFloat(node.GPUCost, 64)
  636. if math.IsNaN(gpuCost) || math.IsInf(gpuCost, 0) {
  637. gpuCost, _ = strconv.ParseFloat(cfg.GPU, 64)
  638. if math.IsNaN(gpuCost) || math.IsInf(gpuCost, 0) {
  639. gpuCost = 0
  640. }
  641. }
  642. nodeType := node.InstanceType
  643. nodeRegion := node.Region
  644. totalCost := cpu*cpuCost + ramCost*(ram/1024/1024/1024) + gpu*gpuCost
  645. a.CPUPriceRecorder.WithLabelValues(nodeName, nodeName, nodeType, nodeRegion, node.ProviderID).Set(cpuCost)
  646. a.RAMPriceRecorder.WithLabelValues(nodeName, nodeName, nodeType, nodeRegion, node.ProviderID).Set(ramCost)
  647. a.GPUPriceRecorder.WithLabelValues(nodeName, nodeName, nodeType, nodeRegion, node.ProviderID).Set(gpuCost)
  648. a.NodeTotalPriceRecorder.WithLabelValues(nodeName, nodeName, nodeType, nodeRegion, node.ProviderID).Set(totalCost)
  649. labelKey := getKeyFromLabelStrings(nodeName, nodeName, nodeType, nodeRegion, node.ProviderID)
  650. nodeSeen[labelKey] = true
  651. }
  652. for _, costs := range data {
  653. nodeName := costs.NodeName
  654. namespace := costs.Namespace
  655. podName := costs.PodName
  656. containerName := costs.Name
  657. if costs.PVCData != nil {
  658. for _, pvc := range costs.PVCData {
  659. if pvc.Volume != nil {
  660. a.PVAllocationRecorder.WithLabelValues(namespace, podName, pvc.Claim, pvc.VolumeName).Set(pvc.Values[0].Value)
  661. labelKey := getKeyFromLabelStrings(namespace, podName, pvc.Claim, pvc.VolumeName)
  662. pvcSeen[labelKey] = true
  663. }
  664. }
  665. }
  666. if len(costs.RAMAllocation) > 0 {
  667. a.RAMAllocationRecorder.WithLabelValues(namespace, podName, containerName, nodeName, nodeName).Set(costs.RAMAllocation[0].Value)
  668. }
  669. if len(costs.CPUAllocation) > 0 {
  670. a.CPUAllocationRecorder.WithLabelValues(namespace, podName, containerName, nodeName, nodeName).Set(costs.CPUAllocation[0].Value)
  671. }
  672. if len(costs.GPUReq) > 0 {
  673. // allocation here is set to the request because shared GPU usage not yet supported.
  674. a.GPUAllocationRecorder.WithLabelValues(namespace, podName, containerName, nodeName, nodeName).Set(costs.GPUReq[0].Value)
  675. }
  676. labelKey := getKeyFromLabelStrings(namespace, podName, containerName, nodeName, nodeName)
  677. if podStatus[podName] == v1.PodRunning { // Only report data for current pods
  678. containerSeen[labelKey] = true
  679. } else {
  680. containerSeen[labelKey] = false
  681. }
  682. storageClasses := a.Model.Cache.GetAllStorageClasses()
  683. storageClassMap := make(map[string]map[string]string)
  684. for _, storageClass := range storageClasses {
  685. params := storageClass.Parameters
  686. storageClassMap[storageClass.ObjectMeta.Name] = params
  687. if storageClass.GetAnnotations()["storageclass.kubernetes.io/is-default-class"] == "true" || storageClass.GetAnnotations()["storageclass.beta.kubernetes.io/is-default-class"] == "true" {
  688. storageClassMap["default"] = params
  689. storageClassMap[""] = params
  690. }
  691. }
  692. pvs := a.Model.Cache.GetAllPersistentVolumes()
  693. for _, pv := range pvs {
  694. parameters, ok := storageClassMap[pv.Spec.StorageClassName]
  695. if !ok {
  696. klog.V(4).Infof("Unable to find parameters for storage class \"%s\". Does pv \"%s\" have a storageClassName?", pv.Spec.StorageClassName, pv.Name)
  697. }
  698. var region string
  699. if r, ok := pv.Labels[v1.LabelZoneRegion]; ok {
  700. region = r
  701. } else {
  702. region = defaultRegion
  703. }
  704. cacPv := &costAnalyzerCloud.PV{
  705. Class: pv.Spec.StorageClassName,
  706. Region: region,
  707. Parameters: parameters,
  708. }
  709. GetPVCost(cacPv, pv, a.Cloud, region)
  710. c, _ := strconv.ParseFloat(cacPv.Cost, 64)
  711. a.PersistentVolumePriceRecorder.WithLabelValues(pv.Name, pv.Name).Set(c)
  712. labelKey := getKeyFromLabelStrings(pv.Name, pv.Name)
  713. pvSeen[labelKey] = true
  714. }
  715. }
  716. for labelString, seen := range nodeSeen {
  717. if !seen {
  718. klog.Infof("Removing %s from nodes", labelString)
  719. labels := getLabelStringsFromKey(labelString)
  720. ok := a.NodeTotalPriceRecorder.DeleteLabelValues(labels...)
  721. if ok {
  722. klog.Infof("removed %s from totalprice", labelString)
  723. } else {
  724. klog.Infof("FAILURE TO REMOVE %s from totalprice", labelString)
  725. }
  726. ok = a.CPUPriceRecorder.DeleteLabelValues(labels...)
  727. if ok {
  728. klog.Infof("removed %s from cpuprice", labelString)
  729. } else {
  730. klog.Infof("FAILURE TO REMOVE %s from cpuprice", labelString)
  731. }
  732. ok = a.GPUPriceRecorder.DeleteLabelValues(labels...)
  733. if ok {
  734. klog.Infof("removed %s from gpuprice", labelString)
  735. } else {
  736. klog.Infof("FAILURE TO REMOVE %s from gpuprice", labelString)
  737. }
  738. ok = a.RAMPriceRecorder.DeleteLabelValues(labels...)
  739. if ok {
  740. klog.Infof("removed %s from ramprice", labelString)
  741. } else {
  742. klog.Infof("FAILURE TO REMOVE %s from ramprice", labelString)
  743. }
  744. delete(nodeSeen, labelString)
  745. }
  746. nodeSeen[labelString] = false
  747. }
  748. for labelString, seen := range containerSeen {
  749. if !seen {
  750. labels := getLabelStringsFromKey(labelString)
  751. a.RAMAllocationRecorder.DeleteLabelValues(labels...)
  752. a.CPUAllocationRecorder.DeleteLabelValues(labels...)
  753. a.GPUAllocationRecorder.DeleteLabelValues(labels...)
  754. delete(containerSeen, labelString)
  755. }
  756. containerSeen[labelString] = false
  757. }
  758. for labelString, seen := range pvSeen {
  759. if !seen {
  760. labels := getLabelStringsFromKey(labelString)
  761. a.PersistentVolumePriceRecorder.DeleteLabelValues(labels...)
  762. delete(pvSeen, labelString)
  763. }
  764. pvSeen[labelString] = false
  765. }
  766. for labelString, seen := range pvcSeen {
  767. if !seen {
  768. labels := getLabelStringsFromKey(labelString)
  769. a.PVAllocationRecorder.DeleteLabelValues(labels...)
  770. delete(pvcSeen, labelString)
  771. }
  772. pvcSeen[labelString] = false
  773. }
  774. time.Sleep(time.Minute)
  775. }
  776. }()
  777. }
  778. // Creates a new ClusterManager instance using a boltdb storage. If that fails,
  779. // then we fall back to a memory-only storage.
  780. func newClusterManager() *cm.ClusterManager {
  781. clustersConfigFile := "/var/configs/clusters/default-clusters.yaml"
  782. // Return a memory-backed cluster manager populated by configmap
  783. return cm.NewConfiguredClusterManager(cm.NewMapDBClusterStorage(), clustersConfigFile)
  784. // NOTE: The following should be used with a persistent disk store. Since the
  785. // NOTE: configmap approach is currently the "persistent" source (entries are read-only
  786. // NOTE: on the backend), we don't currently need to store on disk.
  787. /*
  788. path := env.GetConfigPath()
  789. db, err := bolt.Open(path+"costmodel.db", 0600, nil)
  790. if err != nil {
  791. klog.V(1).Infof("[Error] Failed to create costmodel.db: %s", err.Error())
  792. return cm.NewConfiguredClusterManager(cm.NewMapDBClusterStorage(), clustersConfigFile)
  793. }
  794. store, err := cm.NewBoltDBClusterStorage("clusters", db)
  795. if err != nil {
  796. klog.V(1).Infof("[Error] Failed to Create Cluster Storage: %s", err.Error())
  797. return cm.NewConfiguredClusterManager(cm.NewMapDBClusterStorage(), clustersConfigFile)
  798. }
  799. return cm.NewConfiguredClusterManager(store, clustersConfigFile)
  800. */
  801. }
  802. type ConfigWatchers struct {
  803. ConfigmapName string
  804. WatchFunc func(string, map[string]string) error
  805. }
  806. // captures the panic event in sentry
  807. func capturePanicEvent(err string, stack string) {
  808. msg := fmt.Sprintf("Panic: %s\nStackTrace: %s\n", err, stack)
  809. sentry.CurrentHub().CaptureEvent(&sentry.Event{
  810. Level: sentry.LevelError,
  811. Message: msg,
  812. })
  813. sentry.Flush(5 * time.Second)
  814. }
  815. // handle any panics reported by the errors package
  816. func handlePanic(p errors.Panic) bool {
  817. err := p.Error
  818. if err != nil {
  819. if err, ok := err.(error); ok {
  820. capturePanicEvent(err.Error(), p.Stack)
  821. }
  822. if err, ok := err.(string); ok {
  823. capturePanicEvent(err, p.Stack)
  824. }
  825. }
  826. // Return true to recover iff the type is http, otherwise allow kubernetes
  827. // to recover.
  828. return p.Type == errors.PanicTypeHTTP
  829. }
  830. func Initialize(additionalConfigWatchers ...ConfigWatchers) {
  831. klog.InitFlags(nil)
  832. flag.Set("v", "3")
  833. flag.Parse()
  834. klog.V(1).Infof("Starting cost-model (git commit \"%s\")", gitCommit)
  835. var err error
  836. if errorReportingEnabled {
  837. err = sentry.Init(sentry.ClientOptions{Release: gitCommit})
  838. if err != nil {
  839. klog.Infof("Failed to initialize sentry for error reporting")
  840. } else {
  841. err = errors.SetPanicHandler(handlePanic)
  842. if err != nil {
  843. klog.Infof("Failed to set panic handler: %s", err)
  844. }
  845. }
  846. }
  847. address := env.GetPrometheusServerEndpoint()
  848. if address == "" {
  849. klog.Fatalf("No address for prometheus set in $%s. Aborting.", env.PrometheusServerEndpointEnvVar)
  850. }
  851. queryConcurrency := env.GetMaxQueryConcurrency()
  852. klog.Infof("Prometheus/Thanos Client Max Concurrency set to %d", queryConcurrency)
  853. var LongTimeoutRoundTripper http.RoundTripper = &http.Transport{ // may be necessary for long prometheus queries. TODO: make this configurable
  854. Proxy: http.ProxyFromEnvironment,
  855. DialContext: (&net.Dialer{
  856. Timeout: 120 * time.Second,
  857. KeepAlive: 120 * time.Second,
  858. }).DialContext,
  859. TLSHandshakeTimeout: 10 * time.Second,
  860. }
  861. pc := prometheusClient.Config{
  862. Address: address,
  863. RoundTripper: LongTimeoutRoundTripper,
  864. }
  865. promCli, _ := prom.NewRateLimitedClient(pc, queryConcurrency, "", "")
  866. m, err := ValidatePrometheus(promCli, false)
  867. if err != nil || m.Running == false {
  868. if err != nil {
  869. klog.Errorf("Failed to query prometheus at %s. Error: %s . Troubleshooting help available at: %s", address, err.Error(), prometheusTroubleshootingEp)
  870. } else if m.Running == false {
  871. klog.Errorf("Prometheus at %s is not running. Troubleshooting help available at: %s", address, prometheusTroubleshootingEp)
  872. }
  873. api := prometheusAPI.NewAPI(promCli)
  874. _, err = api.Config(context.Background())
  875. if err != nil {
  876. klog.Infof("No valid prometheus config file at %s. Error: %s . Troubleshooting help available at: %s. Ignore if using cortex/thanos here.", address, err.Error(), prometheusTroubleshootingEp)
  877. } else {
  878. klog.V(1).Info("Retrieved a prometheus config file from: " + address)
  879. }
  880. } else {
  881. klog.V(1).Info("Success: retrieved the 'up' query against prometheus at: " + address)
  882. }
  883. // Kubernetes API setup
  884. kc, err := rest.InClusterConfig()
  885. if err != nil {
  886. panic(err.Error())
  887. }
  888. kubeClientset, err := kubernetes.NewForConfig(kc)
  889. if err != nil {
  890. panic(err.Error())
  891. }
  892. // Create Kubernetes Cluster Cache + Watchers
  893. k8sCache := clustercache.NewKubernetesClusterCache(kubeClientset)
  894. k8sCache.Run()
  895. cloudProviderKey := env.GetCloudProviderAPIKey()
  896. cloudProvider, err := costAnalyzerCloud.NewProvider(k8sCache, cloudProviderKey)
  897. if err != nil {
  898. panic(err.Error())
  899. }
  900. watchConfigFunc := func(c interface{}) {
  901. conf := c.(*v1.ConfigMap)
  902. if conf.GetName() == "pricing-configs" {
  903. _, err := cloudProvider.UpdateConfigFromConfigMap(conf.Data)
  904. if err != nil {
  905. klog.Infof("ERROR UPDATING %s CONFIG: %s", "pricing-configs", err.Error())
  906. }
  907. }
  908. for _, cw := range additionalConfigWatchers {
  909. if conf.GetName() == cw.ConfigmapName {
  910. err := cw.WatchFunc(conf.GetName(), conf.Data)
  911. if err != nil {
  912. klog.Infof("ERROR UPDATING %s CONFIG: %s", cw.ConfigmapName, err.Error())
  913. }
  914. }
  915. }
  916. }
  917. kubecostNamespace := env.GetKubecostNamespace()
  918. // We need an initial invocation because the init of the cache has happened before we had access to the provider.
  919. configs, err := kubeClientset.CoreV1().ConfigMaps(kubecostNamespace).Get("pricing-configs", metav1.GetOptions{})
  920. if err != nil {
  921. klog.Infof("No %s configmap found at installtime, using existing configs: %s", "pricing-configs", err.Error())
  922. } else {
  923. watchConfigFunc(configs)
  924. }
  925. for _, cw := range additionalConfigWatchers {
  926. configs, err := kubeClientset.CoreV1().ConfigMaps(kubecostNamespace).Get(cw.ConfigmapName, metav1.GetOptions{})
  927. if err != nil {
  928. klog.Infof("No %s configmap found at installtime, using existing configs: %s", cw.ConfigmapName, err.Error())
  929. } else {
  930. watchConfigFunc(configs)
  931. }
  932. }
  933. k8sCache.SetConfigMapUpdateFunc(watchConfigFunc)
  934. // TODO: General Architecture Note: Several passes have been made to modularize a lot of
  935. // TODO: our code, but the router still continues to be the obvious entry point for new \
  936. // TODO: features. We should look to spliting out the actual "router" functionality and
  937. // TODO: implement a builder -> controller for stitching new features and other dependencies.
  938. clusterManager := newClusterManager()
  939. cpuGv := prometheus.NewGaugeVec(prometheus.GaugeOpts{
  940. Name: "node_cpu_hourly_cost",
  941. Help: "node_cpu_hourly_cost hourly cost for each cpu on this node",
  942. }, []string{"instance", "node", "instance_type", "region", "provider_id"})
  943. ramGv := prometheus.NewGaugeVec(prometheus.GaugeOpts{
  944. Name: "node_ram_hourly_cost",
  945. Help: "node_ram_hourly_cost hourly cost for each gb of ram on this node",
  946. }, []string{"instance", "node", "instance_type", "region", "provider_id"})
  947. gpuGv := prometheus.NewGaugeVec(prometheus.GaugeOpts{
  948. Name: "node_gpu_hourly_cost",
  949. Help: "node_gpu_hourly_cost hourly cost for each gpu on this node",
  950. }, []string{"instance", "node", "instance_type", "region", "provider_id"})
  951. totalGv := prometheus.NewGaugeVec(prometheus.GaugeOpts{
  952. Name: "node_total_hourly_cost",
  953. Help: "node_total_hourly_cost Total node cost per hour",
  954. }, []string{"instance", "node", "instance_type", "region", "provider_id"})
  955. pvGv := prometheus.NewGaugeVec(prometheus.GaugeOpts{
  956. Name: "pv_hourly_cost",
  957. Help: "pv_hourly_cost Cost per GB per hour on a persistent disk",
  958. }, []string{"volumename", "persistentvolume"})
  959. RAMAllocation := prometheus.NewGaugeVec(prometheus.GaugeOpts{
  960. Name: "container_memory_allocation_bytes",
  961. Help: "container_memory_allocation_bytes Bytes of RAM used",
  962. }, []string{"namespace", "pod", "container", "instance", "node"})
  963. CPUAllocation := prometheus.NewGaugeVec(prometheus.GaugeOpts{
  964. Name: "container_cpu_allocation",
  965. Help: "container_cpu_allocation Percent of a single CPU used in a minute",
  966. }, []string{"namespace", "pod", "container", "instance", "node"})
  967. GPUAllocation := prometheus.NewGaugeVec(prometheus.GaugeOpts{
  968. Name: "container_gpu_allocation",
  969. Help: "container_gpu_allocation GPU used",
  970. }, []string{"namespace", "pod", "container", "instance", "node"})
  971. PVAllocation := prometheus.NewGaugeVec(prometheus.GaugeOpts{
  972. Name: "pod_pvc_allocation",
  973. Help: "pod_pvc_allocation Bytes used by a PVC attached to a pod",
  974. }, []string{"namespace", "pod", "persistentvolumeclaim", "persistentvolume"})
  975. NetworkZoneEgressRecorder := prometheus.NewGauge(prometheus.GaugeOpts{
  976. Name: "kubecost_network_zone_egress_cost",
  977. Help: "kubecost_network_zone_egress_cost Total cost per GB egress across zones",
  978. })
  979. NetworkRegionEgressRecorder := prometheus.NewGauge(prometheus.GaugeOpts{
  980. Name: "kubecost_network_region_egress_cost",
  981. Help: "kubecost_network_region_egress_cost Total cost per GB egress across regions",
  982. })
  983. NetworkInternetEgressRecorder := prometheus.NewGauge(prometheus.GaugeOpts{
  984. Name: "kubecost_network_internet_egress_cost",
  985. Help: "kubecost_network_internet_egress_cost Total cost per GB of internet egress.",
  986. })
  987. prometheus.MustRegister(cpuGv)
  988. prometheus.MustRegister(ramGv)
  989. prometheus.MustRegister(gpuGv)
  990. prometheus.MustRegister(totalGv)
  991. prometheus.MustRegister(pvGv)
  992. prometheus.MustRegister(RAMAllocation)
  993. prometheus.MustRegister(CPUAllocation)
  994. prometheus.MustRegister(PVAllocation)
  995. prometheus.MustRegister(GPUAllocation)
  996. prometheus.MustRegister(NetworkZoneEgressRecorder, NetworkRegionEgressRecorder, NetworkInternetEgressRecorder)
  997. prometheus.MustRegister(ServiceCollector{
  998. KubeClientSet: kubeClientset,
  999. })
  1000. prometheus.MustRegister(DeploymentCollector{
  1001. KubeClientSet: kubeClientset,
  1002. })
  1003. prometheus.MustRegister(StatefulsetCollector{
  1004. KubeClientSet: kubeClientset,
  1005. })
  1006. // cache responses from model for a default of 5 minutes; clear expired responses every 10 minutes
  1007. outOfClusterCache := cache.New(time.Minute*5, time.Minute*10)
  1008. A = Accesses{
  1009. PrometheusClient: promCli,
  1010. KubeClientSet: kubeClientset,
  1011. ClusterManager: clusterManager,
  1012. Cloud: cloudProvider,
  1013. CPUPriceRecorder: cpuGv,
  1014. RAMPriceRecorder: ramGv,
  1015. GPUPriceRecorder: gpuGv,
  1016. NodeTotalPriceRecorder: totalGv,
  1017. RAMAllocationRecorder: RAMAllocation,
  1018. CPUAllocationRecorder: CPUAllocation,
  1019. GPUAllocationRecorder: GPUAllocation,
  1020. PVAllocationRecorder: PVAllocation,
  1021. NetworkZoneEgressRecorder: NetworkZoneEgressRecorder,
  1022. NetworkRegionEgressRecorder: NetworkRegionEgressRecorder,
  1023. NetworkInternetEgressRecorder: NetworkInternetEgressRecorder,
  1024. PersistentVolumePriceRecorder: pvGv,
  1025. Model: NewCostModel(k8sCache),
  1026. OutOfClusterCache: outOfClusterCache,
  1027. }
  1028. remoteEnabled := env.IsRemoteEnabled()
  1029. if remoteEnabled {
  1030. info, err := cloudProvider.ClusterInfo()
  1031. klog.Infof("Saving cluster with id:'%s', and name:'%s' to durable storage", info["id"], info["name"])
  1032. if err != nil {
  1033. klog.Infof("Error saving cluster id %s", err.Error())
  1034. }
  1035. _, _, err = costAnalyzerCloud.GetOrCreateClusterMeta(info["id"], info["name"])
  1036. if err != nil {
  1037. klog.Infof("Unable to set cluster id '%s' for cluster '%s', %s", info["id"], info["name"], err.Error())
  1038. }
  1039. }
  1040. // Thanos Client
  1041. if thanos.IsEnabled() {
  1042. thanosUrl := thanos.QueryURL()
  1043. if thanosUrl != "" {
  1044. var thanosRT http.RoundTripper = &http.Transport{
  1045. Proxy: http.ProxyFromEnvironment,
  1046. DialContext: (&net.Dialer{
  1047. Timeout: 120 * time.Second,
  1048. KeepAlive: 120 * time.Second,
  1049. }).DialContext,
  1050. TLSHandshakeTimeout: 10 * time.Second,
  1051. }
  1052. thanosConfig := prometheusClient.Config{
  1053. Address: thanosUrl,
  1054. RoundTripper: thanosRT,
  1055. }
  1056. thanosCli, _ := prom.NewRateLimitedClient(thanosConfig, queryConcurrency, multiclusterDBBasicAuthUsername, multiclusterDBBasicAuthPW)
  1057. _, err = ValidatePrometheus(thanosCli, true)
  1058. if err != nil {
  1059. klog.V(1).Infof("[Warning] Failed to query Thanos at %s. Error: %s.", thanosUrl, err.Error())
  1060. A.ThanosClient = thanosCli
  1061. } else {
  1062. klog.V(1).Info("Success: retrieved the 'up' query against Thanos at: " + thanosUrl)
  1063. A.ThanosClient = thanosCli
  1064. }
  1065. } else {
  1066. klog.Infof("Error resolving environment variable: $%s", env.ThanosQueryUrlEnvVar)
  1067. }
  1068. }
  1069. err = A.Cloud.DownloadPricingData()
  1070. if err != nil {
  1071. klog.V(1).Info("Failed to download pricing data: " + err.Error())
  1072. }
  1073. A.recordPrices()
  1074. managerEndpoints := cm.NewClusterManagerEndpoints(A.ClusterManager)
  1075. Router.GET("/costDataModel", A.CostDataModel)
  1076. Router.GET("/costDataModelRange", A.CostDataModelRange)
  1077. Router.GET("/costDataModelRangeLarge", A.CostDataModelRangeLarge)
  1078. Router.GET("/outOfClusterCosts", A.OutOfClusterCostsWithCache)
  1079. Router.GET("/allNodePricing", A.GetAllNodePricing)
  1080. Router.POST("/refreshPricing", A.RefreshPricingData)
  1081. Router.GET("/clusterCostsOverTime", A.ClusterCostsOverTime)
  1082. Router.GET("/clusterCosts", A.ClusterCosts)
  1083. Router.GET("/validatePrometheus", A.GetPrometheusMetadata)
  1084. Router.GET("/managementPlatform", A.ManagementPlatform)
  1085. Router.GET("/clusterInfo", A.ClusterInfo)
  1086. Router.GET("/clusters", managerEndpoints.GetAllClusters)
  1087. Router.GET("/serviceAccountStatus", A.GetServiceAccountStatus)
  1088. Router.PUT("/clusters", managerEndpoints.PutCluster)
  1089. Router.DELETE("/clusters/:id", managerEndpoints.DeleteCluster)
  1090. }