Merge branch 'develop' into bolt/cluster-profile

go.sum

@@ -294,6 +294,7 @@ github.com/prometheus/client_golang v0.9.3-0.20190127221311-3c4408c8b829/go.mod
 github.com/prometheus/client_golang v0.9.3/go.mod h1:/TN21ttK/J9q6uSwhBd54HahCDft0ttaMvbicHlPoso=
 github.com/prometheus/client_golang v1.0.0 h1:vrDKnkGzuGvhNAL56c7DBz29ZL+KxnoR0x7enabFceM=
 github.com/prometheus/client_golang v1.0.0/go.mod h1:db9x61etRT2tGnBNRi70OPL5FsnadC4Ky3P0J6CfImo=
+github.com/prometheus/client_golang v1.7.1 h1:NTGy1Ja9pByO+xAeH/qiWnLrKtr3hJPNjaVUwnjpdpA=
 github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
 github.com/prometheus/client_model v0.0.0-20190115171406-56726106282f h1:BVwpUVJDADN2ufcGik7W992pyps0wZ888b/y9GXcLTU=
 github.com/prometheus/client_model v0.0.0-20190115171406-56726106282f/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=

pkg/cloud/awsprovider.go

@@ -8,7 +8,6 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
-	"log"
 	"net/http"
 	"os"
 	"regexp"
@@ -22,6 +21,7 @@ import (
 	"github.com/kubecost/cost-model/pkg/clustercache"
 	"github.com/kubecost/cost-model/pkg/env"
 	"github.com/kubecost/cost-model/pkg/errors"
+	"github.com/kubecost/cost-model/pkg/log"
 	"github.com/kubecost/cost-model/pkg/util"
 
 	"github.com/aws/aws-sdk-go/aws"
@@ -1138,9 +1138,9 @@ func (a *AWS) SetKeyEnv() error {
 
 	if _, err := os.Stat(path); err != nil {
 		if os.IsNotExist(err) {
-			log.Printf("error: file %s does not exist", path)
+			log.DedupedErrorf(5, "file %s does not exist", path)
 		} else {
-			log.Printf("error: %s", err)
+			log.DedupedErrorf(5, "other file open error: %s", err)
 		}
 		return err
 	}
@@ -1231,7 +1231,7 @@ func (a *AWS) GetAddresses() ([]byte, error) {
 
 	errors := []error{}
 	for err := range errorCh {
-		log.Printf("[Warning]: unable to get addresses: %s", err)
+		log.DedupedWarningf(5, "unable to get addresses: %s", err)
 		errors = append(errors, err)
 	}
 
@@ -1337,7 +1337,7 @@ func (a *AWS) GetDisks() ([]byte, error) {
 
 	errors := []error{}
 	for err := range errorCh {
-		log.Printf("[Warning]: unable to get disks: %s", err)
+		log.DedupedWarningf(5, "unable to get disks: %s", err)
 		errors = append(errors, err)
 	}
 
@@ -1978,7 +1978,7 @@ func parseSpotData(bucket string, prefix string, projectID string, region string
 				continue
 			}
 
-			klog.V(1).Infof("Found spot info for: %s", spot.InstanceID)
+			log.DedupedInfof(5, "Found spot info for: %s", spot.InstanceID)
 			spots[spot.InstanceID] = &spot
 		}
 		gr.Close()

pkg/costmodel/router.go

@@ -45,12 +45,14 @@ const (
 
 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()
+	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()
@@ -960,7 +962,7 @@ func Initialize(additionalConfigWatchers ...ConfigWatchers) {
 		Address:      address,
 		RoundTripper: LongTimeoutRoundTripper,
 	}
-	promCli, _ := prom.NewRateLimitedClient(pc, queryConcurrency)
+	promCli, _ := prom.NewRateLimitedClient(pc, queryConcurrency, "", "")
 
 	m, err := ValidatePrometheus(promCli, false)
 	if err != nil || m.Running == false {
@@ -1175,7 +1177,8 @@ func Initialize(additionalConfigWatchers ...ConfigWatchers) {
 				Address:      thanosUrl,
 				RoundTripper: thanosRT,
 			}
-			thanosCli, _ := prom.NewRateLimitedClient(thanosConfig, queryConcurrency)
+
+			thanosCli, _ := prom.NewRateLimitedClient(thanosConfig, queryConcurrency, multiclusterDBBasicAuthUsername, multiclusterDBBasicAuthPW)
 
 			_, err = ValidatePrometheus(thanosCli, true)
 			if err != nil {

pkg/env/costmodelenv.go

@@ -27,6 +27,9 @@ const (
 	ProductAnalyticsEnabledEnvVar = "PRODUCT_ANALYTICS_ENABLED"
 	ErrorReportingEnabledEnvVar   = "ERROR_REPORTING_ENABLED"
 	ValuesReportingEnabledEnvVar  = "VALUES_REPORTING_ENABLED"
+
+	MultiClusterBasicAuthUsername = "MC_BASIC_AUTH_USERNAME"
+	MultiClusterBasicAuthPassword = "MC_BASIC_AUTH_PW"
 )
 
 // GetAWSAccessKeyID returns the environment variable value for AWSAccessKeyIDEnvVar which represents
@@ -168,3 +171,13 @@ func IsValuesReportingEnabled() bool {
 func GetMaxQueryConcurrency() int {
 	return GetInt(MaxQueryConcurrencyEnvVar, 5)
 }
+
+// GetMultiClusterBasicAuthUsername returns the environemnt variable value for MultiClusterBasicAuthUsername
+func GetMultiClusterBasicAuthUsername() string {
+	return Get(MultiClusterBasicAuthUsername, "")
+}
+
+// GetMultiClusterBasicAuthPassword returns the environemnt variable value for MultiClusterBasicAuthPassword
+func GetMultiClusterBasicAuthPassword() string {
+	return Get(MultiClusterBasicAuthPassword, "")
+}

pkg/prom/prom.go

@@ -11,7 +11,7 @@ import (
 
 // NewRateLimitedClient creates a prometheus client which limits the number of concurrent outbound
 // prometheus requests.
-func NewRateLimitedClient(config prometheus.Config, maxConcurrency int) (prometheus.Client, error) {
+func NewRateLimitedClient(config prometheus.Config, maxConcurrency int, username, password string) (prometheus.Client, error) {
 	c, err := prometheus.NewClient(config)
 	if err != nil {
 		return nil, err
@@ -20,16 +20,20 @@ func NewRateLimitedClient(config prometheus.Config, maxConcurrency int) (prometh
 	limiter := util.NewSemaphore(maxConcurrency)
 
 	return &RateLimitedPrometheusClient{
-		client:  c,
-		limiter: limiter,
+		client:   c,
+		limiter:  limiter,
+		username: username,
+		password: password,
 	}, nil
 }
 
 // Creates a new prometheus client which limits the total number of concurrent outbound requests
 // allowed at a given moment.
 type RateLimitedPrometheusClient struct {
-	client  prometheus.Client
-	limiter *util.Semaphore
+	client   prometheus.Client
+	limiter  *util.Semaphore
+	username string
+	password string
 }
 
 // Passthrough to the prometheus client API
@@ -39,6 +43,9 @@ func (rlpc *RateLimitedPrometheusClient) URL(ep string, args map[string]string)
 
 // Rate limit and passthrough to prometheus client API
 func (rlpc *RateLimitedPrometheusClient) Do(ctx context.Context, req *http.Request) (*http.Response, []byte, prometheus.Warnings, error) {
+	if rlpc.username != "" {
+		req.SetBasicAuth(rlpc.username, rlpc.password)
+	}
 	rlpc.limiter.Acquire()
 	defer rlpc.limiter.Return()
 

pkg/prom/result.go

@@ -19,6 +19,12 @@ type QueryResultsChan chan *QueryResults
 func (qrc QueryResultsChan) Await() []*QueryResult {
 	defer close(qrc)
 	results := <-qrc
+
+	// Possible that the returned results are nil
+	if results == nil {
+		return nil
+	}
+
 	return results.Results
 }