support threadsafe reads

cloud/awsprovider.go

@@ -14,6 +14,7 @@ import (
 	"regexp"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"k8s.io/klog"
@@ -55,6 +56,7 @@ type AWS struct {
 	SpotDataBucket          string
 	SpotDataPrefix          string
 	ProjectID               string
+	DownloadPricingDataLock sync.RWMutex
 	*CustomProvider
 }
 
@@ -347,7 +349,8 @@ func (aws *AWS) isPreemptible(key string) bool {
 
 // DownloadPricingData fetches data from the AWS Pricing API
 func (aws *AWS) DownloadPricingData() error {
-
+	aws.DownloadPricingDataLock.Lock()
+	defer aws.DownloadPricingDataLock.Unlock()
 	c, err := GetDefaultPricingData("aws.json")
 	if err != nil {
 		klog.V(1).Infof("Error downloading default pricing data: %s", err.Error())
@@ -502,6 +505,8 @@ func (aws *AWS) DownloadPricingData() error {
 
 // AllNodePricing returns all the billing data fetched.
 func (aws *AWS) AllNodePricing() (interface{}, error) {
+	aws.DownloadPricingDataLock.RLock()
+	defer aws.DownloadPricingDataLock.RUnlock()
 	return aws.Pricing, nil
 }
 
@@ -555,7 +560,9 @@ func (aws *AWS) createNode(terms *AWSProductTerms, usageType string, k Key) (*No
 
 // NodePricing takes in a key from GetKey and returns a Node object for use in building the cost model.
 func (aws *AWS) NodePricing(k Key) (*Node, error) {
-	//return json.Marshal(aws.Pricing[key])
+	aws.DownloadPricingDataLock.RLock()
+	defer aws.DownloadPricingDataLock.RUnlock()
+
 	key := k.Features()
 	usageType := "ondemand"
 	if aws.isPreemptible(key) {
@@ -566,7 +573,9 @@ func (aws *AWS) NodePricing(k Key) (*Node, error) {
 	if ok {
 		return aws.createNode(terms, usageType, k)
 	} else if _, ok := aws.ValidPricingKeys[key]; ok {
+		aws.DownloadPricingDataLock.RUnlock()
 		err := aws.DownloadPricingData()
+		aws.DownloadPricingDataLock.RLock()
 		if err != nil {
 			return nil, err
 		}

cloud/gcpprovider.go

@@ -13,6 +13,7 @@ import (
 	"regexp"
 	"strconv"
 	"strings"
+	"sync"
 
 	"k8s.io/klog"
 
@@ -42,12 +43,13 @@ func (t userAgentTransport) RoundTrip(req *http.Request) (*http.Response, error)
 
 // GCP implements a provider interface for GCP
 type GCP struct {
-	Pricing            map[string]*GCPPricing
-	Clientset          *kubernetes.Clientset
-	APIKey             string
-	BaseCPUPrice       string
-	ProjectID          string
-	BillingDataDataset string
+	Pricing                 map[string]*GCPPricing
+	Clientset               *kubernetes.Clientset
+	APIKey                  string
+	BaseCPUPrice            string
+	ProjectID               string
+	BillingDataDataset      string
+	DownloadPricingDataLock sync.RWMutex
 	*CustomProvider
 }
 
@@ -583,7 +585,8 @@ func (gcp *GCP) parsePages(inputKeys map[string]Key, pvKeys map[string]PVKey) (m
 
 // DownloadPricingData fetches data from the GCP Pricing API. Requires a key-- a kubecost key is provided for quickstart, but should be replaced by a users.
 func (gcp *GCP) DownloadPricingData() error {
-
+	gcp.DownloadPricingDataLock.Lock()
+	defer gcp.DownloadPricingDataLock.Unlock()
 	c, err := GetDefaultPricingData("gcp.json")
 	if err != nil {
 		klog.V(2).Infof("Error downloading default pricing data: %s", err.Error())
@@ -634,11 +637,12 @@ func (gcp *GCP) DownloadPricingData() error {
 		return err
 	}
 	gcp.Pricing = pages
-
 	return nil
 }
 
 func (gcp *GCP) PVPricing(pvk PVKey) (*PV, error) {
+	gcp.DownloadPricingDataLock.RLock()
+	defer gcp.DownloadPricingDataLock.RUnlock()
 	pricing, ok := gcp.Pricing[pvk.Features()]
 	if !ok {
 		klog.V(2).Infof("Persistent Volume pricing not found for %s", pvk)
@@ -719,11 +723,15 @@ func (gcp *gcpKey) Features() string {
 
 // AllNodePricing returns the GCP pricing objects stored
 func (gcp *GCP) AllNodePricing() (interface{}, error) {
+	gcp.DownloadPricingDataLock.RLock()
+	defer gcp.DownloadPricingDataLock.RUnlock()
 	return gcp.Pricing, nil
 }
 
 // NodePricing returns GCP pricing data for a single node
 func (gcp *GCP) NodePricing(key Key) (*Node, error) {
+	gcp.DownloadPricingDataLock.RLock()
+	defer gcp.DownloadPricingDataLock.RUnlock()
 	if n, ok := gcp.Pricing[key.Features()]; ok {
 		klog.V(4).Infof("Returning pricing for node %s: %+v from SKU %s", key, n.Node, n.Name)
 		n.Node.BaseCPUPrice = gcp.BaseCPUPrice

cloud/provider.go

@@ -10,6 +10,7 @@ import (
 	"os"
 	"reflect"
 	"strings"
+	"sync"
 
 	"k8s.io/klog"
 
@@ -189,10 +190,11 @@ type NodePrice struct {
 }
 
 type CustomProvider struct {
-	Clientset      *kubernetes.Clientset
-	Pricing        map[string]*NodePrice
-	SpotLabel      string
-	SpotLabelValue string
+	Clientset               *kubernetes.Clientset
+	Pricing                 map[string]*NodePrice
+	SpotLabel               string
+	SpotLabelValue          string
+	DownloadPricingDataLock sync.RWMutex
 }
 
 func (*CustomProvider) GetConfig() (*CustomPricing, error) {
@@ -216,10 +218,16 @@ func (*CustomProvider) GetDisks() ([]byte, error) {
 }
 
 func (c *CustomProvider) AllNodePricing() (interface{}, error) {
+	c.DownloadPricingDataLock.RLock()
+	defer c.DownloadPricingDataLock.RUnlock()
+
 	return c.Pricing, nil
 }
 
 func (c *CustomProvider) NodePricing(key Key) (*Node, error) {
+	c.DownloadPricingDataLock.RLock()
+	defer c.DownloadPricingDataLock.RUnlock()
+
 	k := key.Features()
 	if _, ok := c.Pricing[k]; !ok {
 		k = "default"
@@ -231,6 +239,8 @@ func (c *CustomProvider) NodePricing(key Key) (*Node, error) {
 }
 
 func (c *CustomProvider) DownloadPricingData() error {
+	c.DownloadPricingDataLock.Lock()
+	defer c.DownloadPricingDataLock.Unlock()
 
 	if c.Pricing == nil {
 		m := make(map[string]*NodePrice)