| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250 |
- package costmodel
- import (
- "context"
- "encoding/json"
- "flag"
- "fmt"
- "math"
- "net"
- "net/http"
- "reflect"
- "strconv"
- "strings"
- "time"
- "k8s.io/klog"
- "github.com/julienschmidt/httprouter"
- sentry "github.com/getsentry/sentry-go"
- costAnalyzerCloud "github.com/kubecost/cost-model/pkg/cloud"
- "github.com/kubecost/cost-model/pkg/clustercache"
- cm "github.com/kubecost/cost-model/pkg/clustermanager"
- "github.com/kubecost/cost-model/pkg/env"
- "github.com/kubecost/cost-model/pkg/errors"
- "github.com/kubecost/cost-model/pkg/prom"
- "github.com/kubecost/cost-model/pkg/thanos"
- prometheusClient "github.com/prometheus/client_golang/api"
- prometheusAPI "github.com/prometheus/client_golang/api/prometheus/v1"
- v1 "k8s.io/api/core/v1"
- metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
- "github.com/patrickmn/go-cache"
- "github.com/prometheus/client_golang/prometheus"
- "k8s.io/client-go/kubernetes"
- "k8s.io/client-go/rest"
- )
- const (
- prometheusTroubleshootingEp = "http://docs.kubecost.com/custom-prom#troubleshoot"
- RFC3339Milli = "2006-01-02T15:04:05.000Z"
- )
- var (
- // gitCommit is set by the build system
- gitCommit string
- logCollectionEnabled bool = env.IsLogCollectionEnabled()
- productAnalyticsEnabled bool = env.IsProductAnalyticsEnabled()
- errorReportingEnabled bool = env.IsErrorReportingEnabled()
- valuesReportingEnabled bool = env.IsValuesReportingEnabled()
- clusterProfile string = env.GetClusterProfile()
- multiclusterDBBasicAuthUsername string = env.GetMultiClusterBasicAuthUsername()
- multiclusterDBBasicAuthPW string = env.GetMultiClusterBasicAuthPassword()
- )
- var Router = httprouter.New()
- var A Accesses
- type Accesses struct {
- PrometheusClient prometheusClient.Client
- ThanosClient prometheusClient.Client
- KubeClientSet kubernetes.Interface
- ClusterManager *cm.ClusterManager
- Cloud costAnalyzerCloud.Provider
- CPUPriceRecorder *prometheus.GaugeVec
- RAMPriceRecorder *prometheus.GaugeVec
- PersistentVolumePriceRecorder *prometheus.GaugeVec
- GPUPriceRecorder *prometheus.GaugeVec
- NodeTotalPriceRecorder *prometheus.GaugeVec
- RAMAllocationRecorder *prometheus.GaugeVec
- CPUAllocationRecorder *prometheus.GaugeVec
- GPUAllocationRecorder *prometheus.GaugeVec
- PVAllocationRecorder *prometheus.GaugeVec
- NetworkZoneEgressRecorder prometheus.Gauge
- NetworkRegionEgressRecorder prometheus.Gauge
- NetworkInternetEgressRecorder prometheus.Gauge
- ServiceSelectorRecorder *prometheus.GaugeVec
- DeploymentSelectorRecorder *prometheus.GaugeVec
- Model *CostModel
- OutOfClusterCache *cache.Cache
- }
- type DataEnvelope struct {
- Code int `json:"code"`
- Status string `json:"status"`
- Data interface{} `json:"data"`
- Message string `json:"message,omitempty"`
- }
- // 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
- type FilterFunc func(*CostData) (bool, string)
- // FilterCostData allows through only CostData that matches all the given filter functions
- func FilterCostData(data map[string]*CostData, retains []FilterFunc, filters []FilterFunc) (map[string]*CostData, int, map[string]int) {
- result := make(map[string]*CostData)
- filteredEnvironments := make(map[string]int)
- filteredContainers := 0
- DataLoop:
- for key, datum := range data {
- for _, rf := range retains {
- if ok, _ := rf(datum); ok {
- result[key] = datum
- // if any retain function passes, the data is retained and move on
- continue DataLoop
- }
- }
- for _, ff := range filters {
- if ok, environment := ff(datum); !ok {
- if environment != "" {
- filteredEnvironments[environment]++
- }
- filteredContainers++
- // if any filter function check fails, move on to the next datum
- continue DataLoop
- }
- }
- result[key] = datum
- }
- return result, filteredContainers, filteredEnvironments
- }
- func filterFields(fields string, data map[string]*CostData) map[string]CostData {
- fs := strings.Split(fields, ",")
- fmap := make(map[string]bool)
- for _, f := range fs {
- fieldNameLower := strings.ToLower(f) // convert to go struct name by uppercasing first letter
- klog.V(1).Infof("to delete: %s", fieldNameLower)
- fmap[fieldNameLower] = true
- }
- filteredData := make(map[string]CostData)
- for cname, costdata := range data {
- s := reflect.TypeOf(*costdata)
- val := reflect.ValueOf(*costdata)
- costdata2 := CostData{}
- cd2 := reflect.New(reflect.Indirect(reflect.ValueOf(costdata2)).Type()).Elem()
- n := s.NumField()
- for i := 0; i < n; i++ {
- field := s.Field(i)
- value := val.Field(i)
- value2 := cd2.Field(i)
- if _, ok := fmap[strings.ToLower(field.Name)]; !ok {
- value2.Set(reflect.Value(value))
- }
- }
- filteredData[cname] = cd2.Interface().(CostData)
- }
- return filteredData
- }
- func normalizeTimeParam(param string) (string, error) {
- if param == "" {
- return "", fmt.Errorf("invalid time param")
- }
- // convert days to hours
- if param[len(param)-1:] == "d" {
- count := param[:len(param)-1]
- val, err := strconv.ParseInt(count, 10, 64)
- if err != nil {
- return "", err
- }
- val = val * 24
- param = fmt.Sprintf("%dh", val)
- }
- return param, nil
- }
- // writeReportingFlags writes the reporting flags to the cluster info map
- func writeReportingFlags(clusterInfo map[string]string) {
- clusterInfo["logCollection"] = fmt.Sprintf("%t", logCollectionEnabled)
- clusterInfo["productAnalytics"] = fmt.Sprintf("%t", productAnalyticsEnabled)
- clusterInfo["errorReporting"] = fmt.Sprintf("%t", errorReportingEnabled)
- clusterInfo["valuesReporting"] = fmt.Sprintf("%t", valuesReportingEnabled)
- }
- // parsePercentString takes a string of expected format "N%" and returns a floating point 0.0N.
- // If the "%" symbol is missing, it just returns 0.0N. Empty string is interpreted as "0%" and
- // return 0.0.
- func ParsePercentString(percentStr string) (float64, error) {
- if len(percentStr) == 0 {
- return 0.0, nil
- }
- if percentStr[len(percentStr)-1:] == "%" {
- percentStr = percentStr[:len(percentStr)-1]
- }
- discount, err := strconv.ParseFloat(percentStr, 64)
- if err != nil {
- return 0.0, err
- }
- discount *= 0.01
- return discount, nil
- }
- // parseDuration converts a Prometheus-style duration string into a Duration
- func ParseDuration(duration string) (*time.Duration, error) {
- unitStr := duration[len(duration)-1:]
- var unit time.Duration
- switch unitStr {
- case "s":
- unit = time.Second
- case "m":
- unit = time.Minute
- case "h":
- unit = time.Hour
- case "d":
- unit = 24.0 * time.Hour
- default:
- return nil, fmt.Errorf("error parsing duration: %s did not match expected format [0-9+](s|m|d|h)", duration)
- }
- amountStr := duration[:len(duration)-1]
- amount, err := strconv.ParseInt(amountStr, 10, 64)
- if err != nil {
- return nil, fmt.Errorf("error parsing duration: %s did not match expected format [0-9+](s|m|d|h)", duration)
- }
- dur := time.Duration(amount) * unit
- return &dur, nil
- }
- // ParseTimeRange returns a start and end time, respectively, which are converted from
- // a duration and offset, defined as strings with Prometheus-style syntax.
- func ParseTimeRange(duration, offset string) (*time.Time, *time.Time, error) {
- // endTime defaults to the current time, unless an offset is explicity declared,
- // in which case it shifts endTime back by given duration
- endTime := time.Now()
- if offset != "" {
- o, err := ParseDuration(offset)
- if err != nil {
- return nil, nil, fmt.Errorf("error parsing offset (%s): %s", offset, err)
- }
- endTime = endTime.Add(-1 * *o)
- }
- // if duration is defined in terms of days, convert to hours
- // e.g. convert "2d" to "48h"
- durationNorm, err := normalizeTimeParam(duration)
- if err != nil {
- return nil, nil, fmt.Errorf("error parsing duration (%s): %s", duration, err)
- }
- // convert time duration into start and end times, formatted
- // as ISO datetime strings
- dur, err := time.ParseDuration(durationNorm)
- if err != nil {
- return nil, nil, fmt.Errorf("errorf parsing duration (%s): %s", durationNorm, err)
- }
- startTime := endTime.Add(-1 * dur)
- return &startTime, &endTime, nil
- }
- func WrapDataWithMessage(data interface{}, err error, message string) []byte {
- var resp []byte
- if err != nil {
- klog.V(1).Infof("Error returned to client: %s", err.Error())
- resp, _ = json.Marshal(&DataEnvelope{
- Code: http.StatusInternalServerError,
- Status: "error",
- Message: err.Error(),
- Data: data,
- })
- } else {
- resp, _ = json.Marshal(&DataEnvelope{
- Code: http.StatusOK,
- Status: "success",
- Data: data,
- Message: message,
- })
- }
- return resp
- }
- func WrapData(data interface{}, err error) []byte {
- var resp []byte
- if err != nil {
- klog.V(1).Infof("Error returned to client: %s", err.Error())
- resp, _ = json.Marshal(&DataEnvelope{
- Code: http.StatusInternalServerError,
- Status: "error",
- Message: err.Error(),
- Data: data,
- })
- } else {
- resp, _ = json.Marshal(&DataEnvelope{
- Code: http.StatusOK,
- Status: "success",
- Data: data,
- })
- }
- return resp
- }
- // 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.
- func (a *Accesses) RefreshPricingData(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
- w.Header().Set("Content-Type", "application/json")
- w.Header().Set("Access-Control-Allow-Origin", "*")
- err := a.Cloud.DownloadPricingData()
- w.Write(WrapData(nil, err))
- }
- func (a *Accesses) CostDataModel(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
- w.Header().Set("Content-Type", "application/json")
- w.Header().Set("Access-Control-Allow-Origin", "*")
- window := r.URL.Query().Get("timeWindow")
- offset := r.URL.Query().Get("offset")
- fields := r.URL.Query().Get("filterFields")
- namespace := r.URL.Query().Get("namespace")
- if offset != "" {
- offset = "offset " + offset
- }
- data, err := a.Model.ComputeCostData(a.PrometheusClient, a.KubeClientSet, a.Cloud, window, offset, namespace)
- if fields != "" {
- filteredData := filterFields(fields, data)
- w.Write(WrapData(filteredData, err))
- } else {
- w.Write(WrapData(data, err))
- }
- }
- func (a *Accesses) ClusterCosts(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
- w.Header().Set("Content-Type", "application/json")
- w.Header().Set("Access-Control-Allow-Origin", "*")
- window := r.URL.Query().Get("window")
- offset := r.URL.Query().Get("offset")
- data, err := ComputeClusterCosts(a.PrometheusClient, a.Cloud, window, offset, true)
- w.Write(WrapData(data, err))
- }
- func (a *Accesses) ClusterCostsOverTime(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
- w.Header().Set("Content-Type", "application/json")
- w.Header().Set("Access-Control-Allow-Origin", "*")
- start := r.URL.Query().Get("start")
- end := r.URL.Query().Get("end")
- window := r.URL.Query().Get("window")
- offset := r.URL.Query().Get("offset")
- data, err := ClusterCostsOverTime(a.PrometheusClient, a.Cloud, start, end, window, offset)
- w.Write(WrapData(data, err))
- }
- func (a *Accesses) CostDataModelRange(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
- w.Header().Set("Content-Type", "application/json")
- w.Header().Set("Access-Control-Allow-Origin", "*")
- start := r.URL.Query().Get("start")
- end := r.URL.Query().Get("end")
- window := r.URL.Query().Get("window")
- fields := r.URL.Query().Get("filterFields")
- namespace := r.URL.Query().Get("namespace")
- cluster := r.URL.Query().Get("cluster")
- remote := r.URL.Query().Get("remote")
- remoteEnabled := env.IsRemoteEnabled() && remote != "false"
- // Use Thanos Client if it exists (enabled) and remote flag set
- var pClient prometheusClient.Client
- if remote != "false" && a.ThanosClient != nil {
- pClient = a.ThanosClient
- } else {
- pClient = a.PrometheusClient
- }
- resolutionHours := 1.0
- data, err := a.Model.ComputeCostDataRange(pClient, a.KubeClientSet, a.Cloud, start, end, window, resolutionHours, namespace, cluster, remoteEnabled)
- if err != nil {
- w.Write(WrapData(nil, err))
- }
- if fields != "" {
- filteredData := filterFields(fields, data)
- w.Write(WrapData(filteredData, err))
- } else {
- w.Write(WrapData(data, err))
- }
- }
- // CostDataModelRangeLarge is experimental multi-cluster and long-term data storage in SQL support.
- func (a *Accesses) CostDataModelRangeLarge(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
- w.Header().Set("Content-Type", "application/json")
- w.Header().Set("Access-Control-Allow-Origin", "*")
- startString := r.URL.Query().Get("start")
- endString := r.URL.Query().Get("end")
- windowString := r.URL.Query().Get("window")
- var start time.Time
- var end time.Time
- var err error
- if windowString == "" {
- windowString = "1h"
- }
- if startString != "" {
- start, err = time.Parse(RFC3339Milli, startString)
- if err != nil {
- klog.V(1).Infof("Error parsing time " + startString + ". Error: " + err.Error())
- w.Write(WrapData(nil, err))
- }
- } else {
- window, err := time.ParseDuration(windowString)
- if err != nil {
- w.Write(WrapData(nil, fmt.Errorf("Invalid duration '%s'", windowString)))
- }
- start = time.Now().Add(-2 * window)
- }
- if endString != "" {
- end, err = time.Parse(RFC3339Milli, endString)
- if err != nil {
- klog.V(1).Infof("Error parsing time " + endString + ". Error: " + err.Error())
- w.Write(WrapData(nil, err))
- }
- } else {
- end = time.Now()
- }
- remoteLayout := "2006-01-02T15:04:05Z"
- remoteStartStr := start.Format(remoteLayout)
- remoteEndStr := end.Format(remoteLayout)
- klog.V(1).Infof("Using remote database for query from %s to %s with window %s", startString, endString, windowString)
- data, err := CostDataRangeFromSQL("", "", windowString, remoteStartStr, remoteEndStr)
- w.Write(WrapData(data, err))
- }
- func parseAggregations(customAggregation, aggregator, filterType string) (string, []string, string) {
- var key string
- var filter string
- var val []string
- if customAggregation != "" {
- key = customAggregation
- filter = filterType
- val = strings.Split(customAggregation, ",")
- } else {
- aggregations := strings.Split(aggregator, ",")
- for i, agg := range aggregations {
- aggregations[i] = "kubernetes_" + agg
- }
- key = strings.Join(aggregations, ",")
- filter = "kubernetes_" + filterType
- val = aggregations
- }
- return key, val, filter
- }
- func (a *Accesses) OutofClusterCosts(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
- w.Header().Set("Content-Type", "application/json")
- w.Header().Set("Access-Control-Allow-Origin", "*")
- start := r.URL.Query().Get("start")
- end := r.URL.Query().Get("end")
- aggregator := r.URL.Query().Get("aggregator")
- customAggregation := r.URL.Query().Get("customAggregation")
- filterType := r.URL.Query().Get("filterType")
- filterValue := r.URL.Query().Get("filterValue")
- var data []*costAnalyzerCloud.OutOfClusterAllocation
- var err error
- _, aggregations, filter := parseAggregations(customAggregation, aggregator, filterType)
- data, err = a.Cloud.ExternalAllocations(start, end, aggregations, filter, filterValue, false)
- w.Write(WrapData(data, err))
- }
- func (a *Accesses) OutOfClusterCostsWithCache(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
- w.Header().Set("Content-Type", "application/json")
- w.Header().Set("Access-Control-Allow-Origin", "*")
- // start date for which to query costs, inclusive; format YYYY-MM-DD
- start := r.URL.Query().Get("start")
- // end date for which to query costs, inclusive; format YYYY-MM-DD
- end := r.URL.Query().Get("end")
- // aggregator sets the field by which to aggregate; default, prepended by "kubernetes_"
- kubernetesAggregation := r.URL.Query().Get("aggregator")
- // customAggregation allows full customization of aggregator w/o prepending
- customAggregation := r.URL.Query().Get("customAggregation")
- // disableCache, if set to "true", tells this function to recompute and
- // cache the requested data
- disableCache := r.URL.Query().Get("disableCache") == "true"
- // clearCache, if set to "true", tells this function to flush the cache,
- // then recompute and cache the requested data
- clearCache := r.URL.Query().Get("clearCache") == "true"
- filterType := r.URL.Query().Get("filterType")
- filterValue := r.URL.Query().Get("filterValue")
- aggregationkey, aggregation, filter := parseAggregations(customAggregation, kubernetesAggregation, filterType)
- // clear cache prior to checking the cache so that a clearCache=true
- // request always returns a freshly computed value
- if clearCache {
- a.OutOfClusterCache.Flush()
- }
- // attempt to retrieve cost data from cache
- key := fmt.Sprintf(`%s:%s:%s:%s:%s`, start, end, aggregationkey, filter, filterValue)
- if value, found := a.OutOfClusterCache.Get(key); found && !disableCache {
- if data, ok := value.([]*costAnalyzerCloud.OutOfClusterAllocation); ok {
- w.Write(WrapDataWithMessage(data, nil, fmt.Sprintf("out of cluster cache hit: %s", key)))
- return
- }
- klog.Errorf("caching error: failed to type cast data: %s", key)
- }
- data, err := a.Cloud.ExternalAllocations(start, end, aggregation, filter, filterValue, false)
- if err == nil {
- a.OutOfClusterCache.Set(key, data, cache.DefaultExpiration)
- }
- w.Write(WrapDataWithMessage(data, err, fmt.Sprintf("out of cluser cache miss: %s", key)))
- }
- func (p *Accesses) GetAllNodePricing(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
- w.Header().Set("Content-Type", "application/json")
- w.Header().Set("Access-Control-Allow-Origin", "*")
- data, err := p.Cloud.AllNodePricing()
- w.Write(WrapData(data, err))
- }
- func (p *Accesses) GetConfigs(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
- w.Header().Set("Content-Type", "application/json")
- w.Header().Set("Access-Control-Allow-Origin", "*")
- data, err := p.Cloud.GetConfig()
- w.Write(WrapData(data, err))
- }
- func (p *Accesses) UpdateSpotInfoConfigs(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
- w.Header().Set("Content-Type", "application/json")
- w.Header().Set("Access-Control-Allow-Origin", "*")
- data, err := p.Cloud.UpdateConfig(r.Body, costAnalyzerCloud.SpotInfoUpdateType)
- if err != nil {
- w.Write(WrapData(data, err))
- return
- }
- w.Write(WrapData(data, err))
- err = p.Cloud.DownloadPricingData()
- if err != nil {
- klog.V(1).Infof("Error redownloading data on config update: %s", err.Error())
- }
- return
- }
- func (p *Accesses) UpdateAthenaInfoConfigs(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
- w.Header().Set("Content-Type", "application/json")
- w.Header().Set("Access-Control-Allow-Origin", "*")
- data, err := p.Cloud.UpdateConfig(r.Body, costAnalyzerCloud.AthenaInfoUpdateType)
- if err != nil {
- w.Write(WrapData(data, err))
- return
- }
- w.Write(WrapData(data, err))
- return
- }
- func (p *Accesses) UpdateBigQueryInfoConfigs(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
- w.Header().Set("Content-Type", "application/json")
- w.Header().Set("Access-Control-Allow-Origin", "*")
- data, err := p.Cloud.UpdateConfig(r.Body, costAnalyzerCloud.BigqueryUpdateType)
- if err != nil {
- w.Write(WrapData(data, err))
- return
- }
- w.Write(WrapData(data, err))
- return
- }
- func (p *Accesses) UpdateConfigByKey(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
- w.Header().Set("Content-Type", "application/json")
- w.Header().Set("Access-Control-Allow-Origin", "*")
- data, err := p.Cloud.UpdateConfig(r.Body, "")
- if err != nil {
- w.Write(WrapData(data, err))
- return
- }
- w.Write(WrapData(data, err))
- return
- }
- func (p *Accesses) ManagementPlatform(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
- w.Header().Set("Content-Type", "application/json")
- w.Header().Set("Access-Control-Allow-Origin", "*")
- data, err := p.Cloud.GetManagementPlatform()
- if err != nil {
- w.Write(WrapData(data, err))
- return
- }
- w.Write(WrapData(data, err))
- return
- }
- func (p *Accesses) ClusterInfo(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
- w.Header().Set("Content-Type", "application/json")
- w.Header().Set("Access-Control-Allow-Origin", "*")
- data, err := p.Cloud.ClusterInfo()
- kc, ok := p.KubeClientSet.(*kubernetes.Clientset)
- if ok && data != nil {
- v, err := kc.ServerVersion()
- if err != nil {
- klog.Infof("Could not get k8s version info: %s", err.Error())
- } else if v != nil {
- data["version"] = v.Major + "." + v.Minor
- }
- } else {
- klog.Infof("Could not get k8s version info: %s", err.Error())
- }
- // Ensure we create the info object if it doesn't exist
- if data == nil {
- data = make(map[string]string)
- }
- data["clusterProfile"] = clusterProfile
- // Include Product Reporting Flags with Cluster Info
- writeReportingFlags(data)
- // Include Thanos Offset Duration if Applicable
- if thanos.IsEnabled() {
- data["thanosOffset"] = thanos.Offset()
- }
- w.Write(WrapData(data, err))
- }
- func (p *Accesses) GetServiceAccountStatus(w http.ResponseWriter, _ *http.Request, _ httprouter.Params) {
- w.Header().Set("Content-Type", "application/json")
- w.Header().Set("Access-Control-Allow-Origin", "*")
- w.Write(WrapData(A.Cloud.ServiceAccountStatus(), nil))
- }
- func (p *Accesses) GetPrometheusMetadata(w http.ResponseWriter, _ *http.Request, _ httprouter.Params) {
- w.Header().Set("Content-Type", "application/json")
- w.Header().Set("Access-Control-Allow-Origin", "*")
- w.Write(WrapData(ValidatePrometheus(p.PrometheusClient, false)))
- }
- func (a *Accesses) recordPrices() {
- go func() {
- defer errors.HandlePanic()
- containerSeen := make(map[string]bool)
- nodeSeen := make(map[string]bool)
- pvSeen := make(map[string]bool)
- pvcSeen := make(map[string]bool)
- getKeyFromLabelStrings := func(labels ...string) string {
- return strings.Join(labels, ",")
- }
- getLabelStringsFromKey := func(key string) []string {
- return strings.Split(key, ",")
- }
- var defaultRegion string = ""
- nodeList := a.Model.Cache.GetAllNodes()
- if len(nodeList) > 0 {
- defaultRegion = nodeList[0].Labels[v1.LabelZoneRegion]
- }
- for {
- klog.V(4).Info("Recording prices...")
- podlist := a.Model.Cache.GetAllPods()
- podStatus := make(map[string]v1.PodPhase)
- for _, pod := range podlist {
- podStatus[pod.Name] = pod.Status.Phase
- }
- cfg, _ := a.Cloud.GetConfig()
- // Record network pricing at global scope
- networkCosts, err := a.Cloud.NetworkPricing()
- if err != nil {
- klog.V(4).Infof("Failed to retrieve network costs: %s", err.Error())
- } else {
- a.NetworkZoneEgressRecorder.Set(networkCosts.ZoneNetworkEgressCost)
- a.NetworkRegionEgressRecorder.Set(networkCosts.RegionNetworkEgressCost)
- a.NetworkInternetEgressRecorder.Set(networkCosts.InternetNetworkEgressCost)
- }
- data, err := a.Model.ComputeCostData(a.PrometheusClient, a.KubeClientSet, a.Cloud, "2m", "", "")
- if err != nil {
- klog.V(1).Info("Error in price recording: " + err.Error())
- // zero the for loop so the time.Sleep will still work
- data = map[string]*CostData{}
- }
- nodes, err := a.Model.GetNodeCost(a.Cloud)
- for nodeName, node := range nodes {
- // Emit costs, guarding against NaN inputs for custom pricing.
- cpuCost, _ := strconv.ParseFloat(node.VCPUCost, 64)
- if math.IsNaN(cpuCost) || math.IsInf(cpuCost, 0) {
- cpuCost, _ = strconv.ParseFloat(cfg.CPU, 64)
- if math.IsNaN(cpuCost) || math.IsInf(cpuCost, 0) {
- cpuCost = 0
- }
- }
- cpu, _ := strconv.ParseFloat(node.VCPU, 64)
- if math.IsNaN(cpu) || math.IsInf(cpu, 0) {
- cpu = 1 // Assume 1 CPU
- }
- ramCost, _ := strconv.ParseFloat(node.RAMCost, 64)
- if math.IsNaN(ramCost) || math.IsInf(ramCost, 0) {
- ramCost, _ = strconv.ParseFloat(cfg.RAM, 64)
- if math.IsNaN(ramCost) || math.IsInf(ramCost, 0) {
- ramCost = 0
- }
- }
- ram, _ := strconv.ParseFloat(node.RAMBytes, 64)
- if math.IsNaN(ram) || math.IsInf(ram, 0) {
- ram = 0
- }
- gpu, _ := strconv.ParseFloat(node.GPU, 64)
- if math.IsNaN(gpu) || math.IsInf(gpu, 0) {
- gpu = 0
- }
- gpuCost, _ := strconv.ParseFloat(node.GPUCost, 64)
- if math.IsNaN(gpuCost) || math.IsInf(gpuCost, 0) {
- gpuCost, _ = strconv.ParseFloat(cfg.GPU, 64)
- if math.IsNaN(gpuCost) || math.IsInf(gpuCost, 0) {
- gpuCost = 0
- }
- }
- nodeType := node.InstanceType
- nodeRegion := node.Region
- totalCost := cpu*cpuCost + ramCost*(ram/1024/1024/1024) + gpu*gpuCost
- a.CPUPriceRecorder.WithLabelValues(nodeName, nodeName, nodeType, nodeRegion, node.ProviderID).Set(cpuCost)
- a.RAMPriceRecorder.WithLabelValues(nodeName, nodeName, nodeType, nodeRegion, node.ProviderID).Set(ramCost)
- a.GPUPriceRecorder.WithLabelValues(nodeName, nodeName, nodeType, nodeRegion, node.ProviderID).Set(gpuCost)
- a.NodeTotalPriceRecorder.WithLabelValues(nodeName, nodeName, nodeType, nodeRegion, node.ProviderID).Set(totalCost)
- labelKey := getKeyFromLabelStrings(nodeName, nodeName, nodeType, nodeRegion, node.ProviderID)
- nodeSeen[labelKey] = true
- }
- for _, costs := range data {
- nodeName := costs.NodeName
- namespace := costs.Namespace
- podName := costs.PodName
- containerName := costs.Name
- if costs.PVCData != nil {
- for _, pvc := range costs.PVCData {
- if pvc.Volume != nil {
- a.PVAllocationRecorder.WithLabelValues(namespace, podName, pvc.Claim, pvc.VolumeName).Set(pvc.Values[0].Value)
- labelKey := getKeyFromLabelStrings(namespace, podName, pvc.Claim, pvc.VolumeName)
- pvcSeen[labelKey] = true
- }
- }
- }
- if len(costs.RAMAllocation) > 0 {
- a.RAMAllocationRecorder.WithLabelValues(namespace, podName, containerName, nodeName, nodeName).Set(costs.RAMAllocation[0].Value)
- }
- if len(costs.CPUAllocation) > 0 {
- a.CPUAllocationRecorder.WithLabelValues(namespace, podName, containerName, nodeName, nodeName).Set(costs.CPUAllocation[0].Value)
- }
- if len(costs.GPUReq) > 0 {
- // allocation here is set to the request because shared GPU usage not yet supported.
- a.GPUAllocationRecorder.WithLabelValues(namespace, podName, containerName, nodeName, nodeName).Set(costs.GPUReq[0].Value)
- }
- labelKey := getKeyFromLabelStrings(namespace, podName, containerName, nodeName, nodeName)
- if podStatus[podName] == v1.PodRunning { // Only report data for current pods
- containerSeen[labelKey] = true
- } else {
- containerSeen[labelKey] = false
- }
- storageClasses := a.Model.Cache.GetAllStorageClasses()
- storageClassMap := make(map[string]map[string]string)
- for _, storageClass := range storageClasses {
- params := storageClass.Parameters
- storageClassMap[storageClass.ObjectMeta.Name] = params
- if storageClass.GetAnnotations()["storageclass.kubernetes.io/is-default-class"] == "true" || storageClass.GetAnnotations()["storageclass.beta.kubernetes.io/is-default-class"] == "true" {
- storageClassMap["default"] = params
- storageClassMap[""] = params
- }
- }
- pvs := a.Model.Cache.GetAllPersistentVolumes()
- for _, pv := range pvs {
- parameters, ok := storageClassMap[pv.Spec.StorageClassName]
- if !ok {
- klog.V(4).Infof("Unable to find parameters for storage class \"%s\". Does pv \"%s\" have a storageClassName?", pv.Spec.StorageClassName, pv.Name)
- }
- var region string
- if r, ok := pv.Labels[v1.LabelZoneRegion]; ok {
- region = r
- } else {
- region = defaultRegion
- }
- cacPv := &costAnalyzerCloud.PV{
- Class: pv.Spec.StorageClassName,
- Region: region,
- Parameters: parameters,
- }
- GetPVCost(cacPv, pv, a.Cloud, region)
- c, _ := strconv.ParseFloat(cacPv.Cost, 64)
- a.PersistentVolumePriceRecorder.WithLabelValues(pv.Name, pv.Name).Set(c)
- labelKey := getKeyFromLabelStrings(pv.Name, pv.Name)
- pvSeen[labelKey] = true
- }
- }
- for labelString, seen := range nodeSeen {
- if !seen {
- klog.Infof("Removing %s from nodes", labelString)
- labels := getLabelStringsFromKey(labelString)
- ok := a.NodeTotalPriceRecorder.DeleteLabelValues(labels...)
- if ok {
- klog.Infof("removed %s from totalprice", labelString)
- } else {
- klog.Infof("FAILURE TO REMOVE %s from totalprice", labelString)
- }
- ok = a.CPUPriceRecorder.DeleteLabelValues(labels...)
- if ok {
- klog.Infof("removed %s from cpuprice", labelString)
- } else {
- klog.Infof("FAILURE TO REMOVE %s from cpuprice", labelString)
- }
- ok = a.GPUPriceRecorder.DeleteLabelValues(labels...)
- if ok {
- klog.Infof("removed %s from gpuprice", labelString)
- } else {
- klog.Infof("FAILURE TO REMOVE %s from gpuprice", labelString)
- }
- ok = a.RAMPriceRecorder.DeleteLabelValues(labels...)
- if ok {
- klog.Infof("removed %s from ramprice", labelString)
- } else {
- klog.Infof("FAILURE TO REMOVE %s from ramprice", labelString)
- }
- delete(nodeSeen, labelString)
- }
- nodeSeen[labelString] = false
- }
- for labelString, seen := range containerSeen {
- if !seen {
- labels := getLabelStringsFromKey(labelString)
- a.RAMAllocationRecorder.DeleteLabelValues(labels...)
- a.CPUAllocationRecorder.DeleteLabelValues(labels...)
- a.GPUAllocationRecorder.DeleteLabelValues(labels...)
- delete(containerSeen, labelString)
- }
- containerSeen[labelString] = false
- }
- for labelString, seen := range pvSeen {
- if !seen {
- labels := getLabelStringsFromKey(labelString)
- a.PersistentVolumePriceRecorder.DeleteLabelValues(labels...)
- delete(pvSeen, labelString)
- }
- pvSeen[labelString] = false
- }
- for labelString, seen := range pvcSeen {
- if !seen {
- labels := getLabelStringsFromKey(labelString)
- a.PVAllocationRecorder.DeleteLabelValues(labels...)
- delete(pvcSeen, labelString)
- }
- pvcSeen[labelString] = false
- }
- time.Sleep(time.Minute)
- }
- }()
- }
- // Creates a new ClusterManager instance using a boltdb storage. If that fails,
- // then we fall back to a memory-only storage.
- func newClusterManager() *cm.ClusterManager {
- clustersConfigFile := "/var/configs/clusters/default-clusters.yaml"
- // Return a memory-backed cluster manager populated by configmap
- return cm.NewConfiguredClusterManager(cm.NewMapDBClusterStorage(), clustersConfigFile)
- // NOTE: The following should be used with a persistent disk store. Since the
- // NOTE: configmap approach is currently the "persistent" source (entries are read-only
- // NOTE: on the backend), we don't currently need to store on disk.
- /*
- path := env.GetConfigPath()
- db, err := bolt.Open(path+"costmodel.db", 0600, nil)
- if err != nil {
- klog.V(1).Infof("[Error] Failed to create costmodel.db: %s", err.Error())
- return cm.NewConfiguredClusterManager(cm.NewMapDBClusterStorage(), clustersConfigFile)
- }
- store, err := cm.NewBoltDBClusterStorage("clusters", db)
- if err != nil {
- klog.V(1).Infof("[Error] Failed to Create Cluster Storage: %s", err.Error())
- return cm.NewConfiguredClusterManager(cm.NewMapDBClusterStorage(), clustersConfigFile)
- }
- return cm.NewConfiguredClusterManager(store, clustersConfigFile)
- */
- }
- type ConfigWatchers struct {
- ConfigmapName string
- WatchFunc func(string, map[string]string) error
- }
- // captures the panic event in sentry
- func capturePanicEvent(err string, stack string) {
- msg := fmt.Sprintf("Panic: %s\nStackTrace: %s\n", err, stack)
- sentry.CurrentHub().CaptureEvent(&sentry.Event{
- Level: sentry.LevelError,
- Message: msg,
- })
- sentry.Flush(5 * time.Second)
- }
- // handle any panics reported by the errors package
- func handlePanic(p errors.Panic) bool {
- err := p.Error
- if err != nil {
- if err, ok := err.(error); ok {
- capturePanicEvent(err.Error(), p.Stack)
- }
- if err, ok := err.(string); ok {
- capturePanicEvent(err, p.Stack)
- }
- }
- // Return true to recover iff the type is http, otherwise allow kubernetes
- // to recover.
- return p.Type == errors.PanicTypeHTTP
- }
- func Initialize(additionalConfigWatchers ...ConfigWatchers) {
- klog.InitFlags(nil)
- flag.Set("v", "3")
- flag.Parse()
- klog.V(1).Infof("Starting cost-model (git commit \"%s\")", gitCommit)
- var err error
- if errorReportingEnabled {
- err = sentry.Init(sentry.ClientOptions{Release: gitCommit})
- if err != nil {
- klog.Infof("Failed to initialize sentry for error reporting")
- } else {
- err = errors.SetPanicHandler(handlePanic)
- if err != nil {
- klog.Infof("Failed to set panic handler: %s", err)
- }
- }
- }
- address := env.GetPrometheusServerEndpoint()
- if address == "" {
- klog.Fatalf("No address for prometheus set in $%s. Aborting.", env.PrometheusServerEndpointEnvVar)
- }
- queryConcurrency := env.GetMaxQueryConcurrency()
- klog.Infof("Prometheus/Thanos Client Max Concurrency set to %d", queryConcurrency)
- var LongTimeoutRoundTripper http.RoundTripper = &http.Transport{ // may be necessary for long prometheus queries. TODO: make this configurable
- Proxy: http.ProxyFromEnvironment,
- DialContext: (&net.Dialer{
- Timeout: 120 * time.Second,
- KeepAlive: 120 * time.Second,
- }).DialContext,
- TLSHandshakeTimeout: 10 * time.Second,
- }
- pc := prometheusClient.Config{
- Address: address,
- RoundTripper: LongTimeoutRoundTripper,
- }
- promCli, _ := prom.NewRateLimitedClient(pc, queryConcurrency, "", "")
- m, err := ValidatePrometheus(promCli, false)
- if err != nil || m.Running == false {
- if err != nil {
- klog.Errorf("Failed to query prometheus at %s. Error: %s . Troubleshooting help available at: %s", address, err.Error(), prometheusTroubleshootingEp)
- } else if m.Running == false {
- klog.Errorf("Prometheus at %s is not running. Troubleshooting help available at: %s", address, prometheusTroubleshootingEp)
- }
- api := prometheusAPI.NewAPI(promCli)
- _, err = api.Config(context.Background())
- if err != nil {
- 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)
- } else {
- klog.V(1).Info("Retrieved a prometheus config file from: " + address)
- }
- } else {
- klog.V(1).Info("Success: retrieved the 'up' query against prometheus at: " + address)
- }
- // Kubernetes API setup
- kc, err := rest.InClusterConfig()
- if err != nil {
- panic(err.Error())
- }
- kubeClientset, err := kubernetes.NewForConfig(kc)
- if err != nil {
- panic(err.Error())
- }
- // Create Kubernetes Cluster Cache + Watchers
- k8sCache := clustercache.NewKubernetesClusterCache(kubeClientset)
- k8sCache.Run()
- cloudProviderKey := env.GetCloudProviderAPIKey()
- cloudProvider, err := costAnalyzerCloud.NewProvider(k8sCache, cloudProviderKey)
- if err != nil {
- panic(err.Error())
- }
- watchConfigFunc := func(c interface{}) {
- conf := c.(*v1.ConfigMap)
- if conf.GetName() == "pricing-configs" {
- _, err := cloudProvider.UpdateConfigFromConfigMap(conf.Data)
- if err != nil {
- klog.Infof("ERROR UPDATING %s CONFIG: %s", "pricing-configs", err.Error())
- }
- }
- for _, cw := range additionalConfigWatchers {
- if conf.GetName() == cw.ConfigmapName {
- err := cw.WatchFunc(conf.GetName(), conf.Data)
- if err != nil {
- klog.Infof("ERROR UPDATING %s CONFIG: %s", cw.ConfigmapName, err.Error())
- }
- }
- }
- }
- kubecostNamespace := env.GetKubecostNamespace()
- // We need an initial invocation because the init of the cache has happened before we had access to the provider.
- configs, err := kubeClientset.CoreV1().ConfigMaps(kubecostNamespace).Get("pricing-configs", metav1.GetOptions{})
- if err != nil {
- klog.Infof("No %s configmap found at installtime, using existing configs: %s", "pricing-configs", err.Error())
- } else {
- watchConfigFunc(configs)
- }
- for _, cw := range additionalConfigWatchers {
- configs, err := kubeClientset.CoreV1().ConfigMaps(kubecostNamespace).Get(cw.ConfigmapName, metav1.GetOptions{})
- if err != nil {
- klog.Infof("No %s configmap found at installtime, using existing configs: %s", cw.ConfigmapName, err.Error())
- } else {
- watchConfigFunc(configs)
- }
- }
- k8sCache.SetConfigMapUpdateFunc(watchConfigFunc)
- // TODO: General Architecture Note: Several passes have been made to modularize a lot of
- // TODO: our code, but the router still continues to be the obvious entry point for new \
- // TODO: features. We should look to spliting out the actual "router" functionality and
- // TODO: implement a builder -> controller for stitching new features and other dependencies.
- clusterManager := newClusterManager()
- cpuGv := prometheus.NewGaugeVec(prometheus.GaugeOpts{
- Name: "node_cpu_hourly_cost",
- Help: "node_cpu_hourly_cost hourly cost for each cpu on this node",
- }, []string{"instance", "node", "instance_type", "region", "provider_id"})
- ramGv := prometheus.NewGaugeVec(prometheus.GaugeOpts{
- Name: "node_ram_hourly_cost",
- Help: "node_ram_hourly_cost hourly cost for each gb of ram on this node",
- }, []string{"instance", "node", "instance_type", "region", "provider_id"})
- gpuGv := prometheus.NewGaugeVec(prometheus.GaugeOpts{
- Name: "node_gpu_hourly_cost",
- Help: "node_gpu_hourly_cost hourly cost for each gpu on this node",
- }, []string{"instance", "node", "instance_type", "region", "provider_id"})
- totalGv := prometheus.NewGaugeVec(prometheus.GaugeOpts{
- Name: "node_total_hourly_cost",
- Help: "node_total_hourly_cost Total node cost per hour",
- }, []string{"instance", "node", "instance_type", "region", "provider_id"})
- pvGv := prometheus.NewGaugeVec(prometheus.GaugeOpts{
- Name: "pv_hourly_cost",
- Help: "pv_hourly_cost Cost per GB per hour on a persistent disk",
- }, []string{"volumename", "persistentvolume"})
- RAMAllocation := prometheus.NewGaugeVec(prometheus.GaugeOpts{
- Name: "container_memory_allocation_bytes",
- Help: "container_memory_allocation_bytes Bytes of RAM used",
- }, []string{"namespace", "pod", "container", "instance", "node"})
- CPUAllocation := prometheus.NewGaugeVec(prometheus.GaugeOpts{
- Name: "container_cpu_allocation",
- Help: "container_cpu_allocation Percent of a single CPU used in a minute",
- }, []string{"namespace", "pod", "container", "instance", "node"})
- GPUAllocation := prometheus.NewGaugeVec(prometheus.GaugeOpts{
- Name: "container_gpu_allocation",
- Help: "container_gpu_allocation GPU used",
- }, []string{"namespace", "pod", "container", "instance", "node"})
- PVAllocation := prometheus.NewGaugeVec(prometheus.GaugeOpts{
- Name: "pod_pvc_allocation",
- Help: "pod_pvc_allocation Bytes used by a PVC attached to a pod",
- }, []string{"namespace", "pod", "persistentvolumeclaim", "persistentvolume"})
- NetworkZoneEgressRecorder := prometheus.NewGauge(prometheus.GaugeOpts{
- Name: "kubecost_network_zone_egress_cost",
- Help: "kubecost_network_zone_egress_cost Total cost per GB egress across zones",
- })
- NetworkRegionEgressRecorder := prometheus.NewGauge(prometheus.GaugeOpts{
- Name: "kubecost_network_region_egress_cost",
- Help: "kubecost_network_region_egress_cost Total cost per GB egress across regions",
- })
- NetworkInternetEgressRecorder := prometheus.NewGauge(prometheus.GaugeOpts{
- Name: "kubecost_network_internet_egress_cost",
- Help: "kubecost_network_internet_egress_cost Total cost per GB of internet egress.",
- })
- prometheus.MustRegister(cpuGv)
- prometheus.MustRegister(ramGv)
- prometheus.MustRegister(gpuGv)
- prometheus.MustRegister(totalGv)
- prometheus.MustRegister(pvGv)
- prometheus.MustRegister(RAMAllocation)
- prometheus.MustRegister(CPUAllocation)
- prometheus.MustRegister(PVAllocation)
- prometheus.MustRegister(GPUAllocation)
- prometheus.MustRegister(NetworkZoneEgressRecorder, NetworkRegionEgressRecorder, NetworkInternetEgressRecorder)
- prometheus.MustRegister(ServiceCollector{
- KubeClientSet: kubeClientset,
- })
- prometheus.MustRegister(DeploymentCollector{
- KubeClientSet: kubeClientset,
- })
- prometheus.MustRegister(StatefulsetCollector{
- KubeClientSet: kubeClientset,
- })
- // cache responses from model for a default of 5 minutes; clear expired responses every 10 minutes
- outOfClusterCache := cache.New(time.Minute*5, time.Minute*10)
- A = Accesses{
- PrometheusClient: promCli,
- KubeClientSet: kubeClientset,
- ClusterManager: clusterManager,
- Cloud: cloudProvider,
- CPUPriceRecorder: cpuGv,
- RAMPriceRecorder: ramGv,
- GPUPriceRecorder: gpuGv,
- NodeTotalPriceRecorder: totalGv,
- RAMAllocationRecorder: RAMAllocation,
- CPUAllocationRecorder: CPUAllocation,
- GPUAllocationRecorder: GPUAllocation,
- PVAllocationRecorder: PVAllocation,
- NetworkZoneEgressRecorder: NetworkZoneEgressRecorder,
- NetworkRegionEgressRecorder: NetworkRegionEgressRecorder,
- NetworkInternetEgressRecorder: NetworkInternetEgressRecorder,
- PersistentVolumePriceRecorder: pvGv,
- Model: NewCostModel(k8sCache),
- OutOfClusterCache: outOfClusterCache,
- }
- remoteEnabled := env.IsRemoteEnabled()
- if remoteEnabled {
- info, err := cloudProvider.ClusterInfo()
- klog.Infof("Saving cluster with id:'%s', and name:'%s' to durable storage", info["id"], info["name"])
- if err != nil {
- klog.Infof("Error saving cluster id %s", err.Error())
- }
- _, _, err = costAnalyzerCloud.GetOrCreateClusterMeta(info["id"], info["name"])
- if err != nil {
- klog.Infof("Unable to set cluster id '%s' for cluster '%s', %s", info["id"], info["name"], err.Error())
- }
- }
- // Thanos Client
- if thanos.IsEnabled() {
- thanosUrl := thanos.QueryURL()
- if thanosUrl != "" {
- var thanosRT http.RoundTripper = &http.Transport{
- Proxy: http.ProxyFromEnvironment,
- DialContext: (&net.Dialer{
- Timeout: 120 * time.Second,
- KeepAlive: 120 * time.Second,
- }).DialContext,
- TLSHandshakeTimeout: 10 * time.Second,
- }
- thanosConfig := prometheusClient.Config{
- Address: thanosUrl,
- RoundTripper: thanosRT,
- }
- thanosCli, _ := prom.NewRateLimitedClient(thanosConfig, queryConcurrency, multiclusterDBBasicAuthUsername, multiclusterDBBasicAuthPW)
- _, err = ValidatePrometheus(thanosCli, true)
- if err != nil {
- klog.V(1).Infof("[Warning] Failed to query Thanos at %s. Error: %s.", thanosUrl, err.Error())
- A.ThanosClient = thanosCli
- } else {
- klog.V(1).Info("Success: retrieved the 'up' query against Thanos at: " + thanosUrl)
- A.ThanosClient = thanosCli
- }
- } else {
- klog.Infof("Error resolving environment variable: $%s", env.ThanosQueryUrlEnvVar)
- }
- }
- err = A.Cloud.DownloadPricingData()
- if err != nil {
- klog.V(1).Info("Failed to download pricing data: " + err.Error())
- }
- A.recordPrices()
- managerEndpoints := cm.NewClusterManagerEndpoints(A.ClusterManager)
- Router.GET("/costDataModel", A.CostDataModel)
- Router.GET("/costDataModelRange", A.CostDataModelRange)
- Router.GET("/costDataModelRangeLarge", A.CostDataModelRangeLarge)
- Router.GET("/outOfClusterCosts", A.OutOfClusterCostsWithCache)
- Router.GET("/allNodePricing", A.GetAllNodePricing)
- Router.POST("/refreshPricing", A.RefreshPricingData)
- Router.GET("/clusterCostsOverTime", A.ClusterCostsOverTime)
- Router.GET("/clusterCosts", A.ClusterCosts)
- Router.GET("/validatePrometheus", A.GetPrometheusMetadata)
- Router.GET("/managementPlatform", A.ManagementPlatform)
- Router.GET("/clusterInfo", A.ClusterInfo)
- Router.GET("/clusters", managerEndpoints.GetAllClusters)
- Router.GET("/serviceAccountStatus", A.GetServiceAccountStatus)
- Router.PUT("/clusters", managerEndpoints.PutCluster)
- Router.DELETE("/clusters/:id", managerEndpoints.DeleteCluster)
- }
|