Merge branch 'develop' into bolt/envvar-thanos-offset

go.sum

@@ -490,6 +490,7 @@ golang.org/x/tools v0.0.0-20190312170243-e65039ee4138 h1:H3uGjxCR/6Ds0Mjgyp7LMK8
 golang.org/x/tools v0.0.0-20190312170243-e65039ee4138/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20190327201419-c70d86f8b7cf/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20190328211700-ab21143f2384/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/api v0.0.0-20181220000619-583d854617af/go.mod h1:4mhQ8q/RsB7i+udVvVy5NUi08OU8ZlA0gRVgrF7VFY0=
 google.golang.org/api v0.2.0/go.mod h1:IfRCZScioGtypHNTlz3gFk67J8uePVW7uDTBzXuIkhU=

pkg/cloud/csvprovider.go

@@ -15,6 +15,7 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/session"
 	"github.com/aws/aws-sdk-go/service/s3"
+	"github.com/kubecost/cost-model/pkg/log"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/klog"
 
@@ -140,7 +141,7 @@ func (c *CSVProvider) DownloadPricingData() error {
 		c.Pricing = pricing
 		c.PricingPV = pvpricing
 	} else {
-		klog.Infof("[WARNING] No data received from csv")
+		log.DedupedWarningf(5, "No data received from csv at %s", c.CSVLocation)
 	}
 	time.AfterFunc(refreshMinutes*time.Minute, func() { c.DownloadPricingData() })
 	return nil

pkg/cloud/gcpprovider.go

@@ -55,6 +55,7 @@ type GCP struct {
 	ReservedInstances       []*GCPReservedInstance
 	Config                  *ProviderConfig
 	serviceKeyProvided      bool
+	ValidPricingKeys        map[string]bool
 	*CustomProvider
 }
 
@@ -707,6 +708,10 @@ func (gcp *GCP) parsePage(r io.Reader, inputKeys map[string]Key, pvKeys map[stri
 				}
 
 				candidateKeys := []string{}
+				if gcp.ValidPricingKeys == nil {
+					gcp.ValidPricingKeys = make(map[string]bool)
+				}
+
 				for _, region := range product.ServiceRegions {
 					if instanceType == "e2" { // this needs to be done to handle a partial cpu mapping
 						candidateKeys = append(candidateKeys, region+","+"e2micro"+","+usageType)
@@ -723,6 +728,8 @@ func (gcp *GCP) parsePage(r io.Reader, inputKeys map[string]Key, pvKeys map[stri
 					instanceType = strings.Split(candidateKey, ",")[1] // we may have overriden this while generating candidate keys
 					region := strings.Split(candidateKey, ",")[0]
 					candidateKeyGPU := candidateKey + ",gpu"
+					gcp.ValidPricingKeys[candidateKey] = true
+					gcp.ValidPricingKeys[candidateKeyGPU] = true
 					if gpuType != "" {
 						lastRateIndex := len(product.PricingInfo[0].PricingExpression.TieredRates) - 1
 						var nanos float64
@@ -1338,15 +1345,37 @@ func (gcp *GCP) AllNodePricing() (interface{}, error) {
 	return gcp.Pricing, nil
 }
 
-// NodePricing returns GCP pricing data for a single node
-func (gcp *GCP) NodePricing(key Key) (*Node, error) {
+func (gcp *GCP) getPricing(key Key) (*GCPPricing, bool) {
+	gcp.DownloadPricingDataLock.RLock()
+	defer gcp.DownloadPricingDataLock.RUnlock()
+	n, ok := gcp.Pricing[key.Features()]
+	return n, ok
+}
+func (gcp *GCP) isValidPricingKey(key Key) bool {
 	gcp.DownloadPricingDataLock.RLock()
 	defer gcp.DownloadPricingDataLock.RUnlock()
-	if n, ok := gcp.Pricing[key.Features()]; ok {
+	_, ok := gcp.ValidPricingKeys[key.Features()]
+	return ok
+}
+
+// NodePricing returns GCP pricing data for a single node
+func (gcp *GCP) NodePricing(key Key) (*Node, error) {
+	if n, ok := gcp.getPricing(key); ok {
 		klog.V(4).Infof("Returning pricing for node %s: %+v from SKU %s", key, n.Node, n.Name)
 		n.Node.BaseCPUPrice = gcp.BaseCPUPrice
 		return n.Node, nil
+	} else if ok := gcp.isValidPricingKey(key); ok {
+		err := gcp.DownloadPricingData()
+		if err != nil {
+			return nil, fmt.Errorf("Download pricing data failed: %s", err.Error())
+		}
+		if n, ok := gcp.getPricing(key); ok {
+			klog.V(4).Infof("Returning pricing for node %s: %+v from SKU %s", key, n.Node, n.Name)
+			n.Node.BaseCPUPrice = gcp.BaseCPUPrice
+			return n.Node, nil
+		}
+		klog.V(1).Infof("[Warning] no pricing data found for %s: %s", key.Features(), key)
+		return nil, fmt.Errorf("Warning: no pricing data found for %s", key)
 	}
-	klog.V(1).Infof("[Warning] no pricing data found for %s: %s", key.Features(), key)
 	return nil, fmt.Errorf("Warning: no pricing data found for %s", key)
 }

pkg/costmodel/costmodel.go

@@ -1142,7 +1142,7 @@ func (cm *CostModel) GetNodeCost(cp costAnalyzerCloud.Provider) (map[string]*cos
 
 		cnode, err := cp.NodePricing(cp.GetKey(nodeLabels, n))
 		if err != nil {
-			klog.V(1).Infof("[Warning] Error getting node pricing. Error: " + err.Error())
+			log.DedupedWarningf(10, "Error getting node pricing. Error: %s", err.Error())
 			if cnode != nil {
 				nodes[name] = cnode
 				continue

pkg/costmodel/router.go

@@ -49,6 +49,7 @@ var (
 	logCollectionEnabled    bool = env.IsLogCollectionEnabled()
 	productAnalyticsEnabled bool = env.IsProductAnalyticsEnabled()
 	errorReportingEnabled   bool = env.IsErrorReportingEnabled()
+	valuesReportingEnabled  bool = env.IsValuesReportingEnabled()
 )
 
 var Router = httprouter.New()
@@ -147,6 +148,9 @@ func filterFields(fields string, data map[string]*CostData) map[string]CostData
 }
 
 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]
@@ -166,6 +170,7 @@ 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.

pkg/env/costmodelenv.go

@@ -25,6 +25,7 @@ const (
 	LogCollectionEnabledEnvVar    = "LOG_COLLECTION_ENABLED"
 	ProductAnalyticsEnabledEnvVar = "PRODUCT_ANALYTICS_ENABLED"
 	ErrorReportingEnabledEnvVar   = "ERROR_REPORTING_ENABLED"
+	ValuesReportingEnabledEnvVar  = "VALUES_REPORTING_ENABLED"
 )
 
 // GetAWSAccessKeyID returns the environment variable value for AWSAccessKeyIDEnvVar which represents
@@ -151,6 +152,11 @@ func IsErrorReportingEnabled() bool {
 	return GetBool(ErrorReportingEnabledEnvVar, true)
 }
 
+// IsValuesReportingEnabled returns the environment variable value for ValuesReportingEnabledEnvVar
+func IsValuesReportingEnabled() bool {
+	return GetBool(ValuesReportingEnabledEnvVar, true)
+}
+
 // GetMaxQueryConcurrency returns the environment variable value for MaxQueryConcurrencyEnvVar
 func GetMaxQueryConcurrency() int {
 	return GetInt(MaxQueryConcurrencyEnvVar, 5)

pkg/log/log.go

@@ -7,18 +7,68 @@ import (
 	"k8s.io/klog"
 )
 
+var seen = make(map[string]int)
+
 func Errorf(format string, a ...interface{}) {
 	klog.Errorf(fmt.Sprintf("[Error] %s", format), a...)
 }
 
+func DedupedErrorf(logTypeLimit int, format string, a ...interface{}) {
+	timesLogged, ok := seen[format]
+	if !ok {
+		seen[format] = 1
+	} else if timesLogged == logTypeLimit {
+		seen[format]++
+		f := fmt.Sprintf("[Error] %s", format)
+		klog.Errorf("%s seen %d times, suppressing future logs", f, logTypeLimit)
+	} else if timesLogged > logTypeLimit {
+		seen[format]++
+	} else {
+		seen[format]++
+		klog.Errorf(fmt.Sprintf("[Error] %s", format), a...)
+	}
+}
+
 func Warningf(format string, a ...interface{}) {
 	klog.V(2).Infof(fmt.Sprintf("[Warning] %s", format), a...)
 }
 
+func DedupedWarningf(logTypeLimit int, format string, a ...interface{}) {
+	timesLogged, ok := seen[format]
+	if !ok {
+		seen[format] = 1
+	} else if timesLogged == logTypeLimit {
+		seen[format]++
+		f := fmt.Sprintf("[Warning] %s", format)
+		klog.Errorf("%s seen %d times, suppressing future logs", f, logTypeLimit)
+	} else if timesLogged > logTypeLimit {
+		seen[format]++
+	} else {
+		seen[format]++
+		klog.V(2).Infof(fmt.Sprintf("[Warning] %s", format), a...)
+	}
+}
+
 func Infof(format string, a ...interface{}) {
 	klog.V(3).Infof(fmt.Sprintf("[Info] %s", format), a...)
 }
 
+func DedupedInfof(logTypeLimit int, format string, a ...interface{}) {
+	timesLogged, ok := seen[format]
+	if !ok {
+		seen[format] = 1
+	} else if timesLogged == logTypeLimit {
+		seen[format]++
+		f := fmt.Sprintf("[Info] %s", format)
+		klog.Errorf("%s seen %d times, suppressing future logs", f, logTypeLimit)
+	} else if timesLogged > logTypeLimit {
+		seen[format]++
+	} else {
+		seen[format]++
+		klog.V(3).Infof(fmt.Sprintf("[Info] %s", format), a...)
+	}
+}
+
 func Profilef(format string, a ...interface{}) {
 	klog.V(3).Infof(fmt.Sprintf("[Profiler] %s", format), a...)
 }