Merge pull request #182 from kubecost/nikovacevic-caching

Aggregation caching improvements

cloud/awsprovider.go

@@ -328,7 +328,7 @@ func (aws *AWS) UpdateConfig(r io.Reader, updateType string) (*CustomPricing, er
 	path += "aws.json"
 	remoteEnabled := os.Getenv(remoteEnabled)
 	if remoteEnabled == "true" {
-		err = UpdateClusterMeta(os.Getenv(KC_CLUSTER_ID), c.ClusterName)
+		err = UpdateClusterMeta(os.Getenv(clusterIDKey), c.ClusterName)
 		if err != nil {
 			return nil, err
 		}
@@ -825,7 +825,7 @@ func (awsProvider *AWS) ClusterInfo() (map[string]string, error) {
 		m := make(map[string]string)
 		m["name"] = c.ClusterName
 		m["provider"] = "AWS"
-		m["id"] = os.Getenv(KC_CLUSTER_ID)
+		m["id"] = os.Getenv(clusterIDKey)
 		m["remoteReadEnabled"] = strconv.FormatBool(remoteEnabled)
 		return m, nil
 	}
@@ -834,7 +834,7 @@ func (awsProvider *AWS) ClusterInfo() (map[string]string, error) {
 		m := make(map[string]string)
 		m["name"] = clusterName
 		m["provider"] = "AWS"
-		m["id"] = os.Getenv(KC_CLUSTER_ID)
+		m["id"] = os.Getenv(clusterIDKey)
 		m["remoteReadEnabled"] = strconv.FormatBool(remoteEnabled)
 		return m, nil
 	}

cloud/azureprovider.go

@@ -506,7 +506,7 @@ func (az *Azure) ClusterInfo() (map[string]string, error) {
 	}
 	m["provider"] = "azure"
 	m["remoteReadEnabled"] = strconv.FormatBool(remoteEnabled)
-	m["id"] = os.Getenv(KC_CLUSTER_ID)
+	m["id"] = os.Getenv(clusterIDKey)
 	return m, nil
 
 }
@@ -543,7 +543,7 @@ func (az *Azure) UpdateConfig(r io.Reader, updateType string) (*CustomPricing, e
 	}
 	remoteEnabled := os.Getenv(remoteEnabled)
 	if remoteEnabled == "true" {
-		err = UpdateClusterMeta(os.Getenv(KC_CLUSTER_ID), c.ClusterName)
+		err = UpdateClusterMeta(os.Getenv(clusterIDKey), c.ClusterName)
 		if err != nil {
 			return nil, err
 		}

cloud/customprovider.go

@@ -0,0 +1,251 @@
+package cloud
+
+import (
+	"encoding/json"
+	"io"
+	"io/ioutil"
+	"net/url"
+	"os"
+	"strconv"
+	"strings"
+	"sync"
+
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/kubernetes"
+)
+
+type NodePrice struct {
+	CPU string
+	RAM string
+	GPU string
+}
+
+type CustomProvider struct {
+	Clientset               *kubernetes.Clientset
+	Pricing                 map[string]*NodePrice
+	SpotLabel               string
+	SpotLabelValue          string
+	GPULabel                string
+	GPULabelValue           string
+	DownloadPricingDataLock sync.RWMutex
+}
+
+type customProviderKey struct {
+	SpotLabel      string
+	SpotLabelValue string
+	GPULabel       string
+	GPULabelValue  string
+	Labels         map[string]string
+}
+
+func (*CustomProvider) GetLocalStorageQuery() (string, error) {
+	return "", nil
+}
+
+func (*CustomProvider) GetConfig() (*CustomPricing, error) {
+	return GetDefaultPricingData("default.json")
+}
+
+func (*CustomProvider) GetManagementPlatform() (string, error) {
+	return "", nil
+}
+
+func (cp *CustomProvider) UpdateConfig(r io.Reader, updateType string) (*CustomPricing, error) {
+	c, err := GetDefaultPricingData("default.json")
+	if err != nil {
+		return nil, err
+	}
+	path := os.Getenv("CONFIG_PATH")
+	if path == "" {
+		path = "/models/"
+	}
+	a := make(map[string]string)
+	err = json.NewDecoder(r).Decode(&a)
+	if err != nil {
+		return nil, err
+	}
+	for k, v := range a {
+		kUpper := strings.Title(k) // Just so we consistently supply / receive the same values, uppercase the first letter.
+		err := SetCustomPricingField(c, kUpper, v)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	cj, err := json.Marshal(c)
+	if err != nil {
+		return nil, err
+	}
+
+	configPath := path + "default.json"
+	err = ioutil.WriteFile(configPath, cj, 0644)
+	if err != nil {
+		return nil, err
+	}
+	defer cp.DownloadPricingData()
+	return c, nil
+
+}
+
+func (cp *CustomProvider) ClusterInfo() (map[string]string, error) {
+	conf, err := cp.GetConfig()
+	if err != nil {
+		return nil, err
+	}
+	m := make(map[string]string)
+	if conf.ClusterName != "" {
+		m["name"] = conf.ClusterName
+	}
+	m["provider"] = "custom"
+	return m, nil
+}
+
+func (*CustomProvider) AddServiceKey(url.Values) error {
+	return nil
+}
+
+func (*CustomProvider) GetDisks() ([]byte, error) {
+	return nil, nil
+}
+
+func (cp *CustomProvider) AllNodePricing() (interface{}, error) {
+	cp.DownloadPricingDataLock.RLock()
+	defer cp.DownloadPricingDataLock.RUnlock()
+
+	return cp.Pricing, nil
+}
+
+func (cp *CustomProvider) NodePricing(key Key) (*Node, error) {
+	cp.DownloadPricingDataLock.RLock()
+	defer cp.DownloadPricingDataLock.RUnlock()
+
+	k := key.Features()
+	var gpuCount string
+	if _, ok := cp.Pricing[k]; !ok {
+		k = "default"
+	}
+	if key.GPUType() != "" {
+		k += ",gpu"    // TODO: support multiple custom gpu types.
+		gpuCount = "1" // TODO: support more than one gpu.
+	}
+
+	return &Node{
+		VCPUCost: cp.Pricing[k].CPU,
+		RAMCost:  cp.Pricing[k].RAM,
+		GPUCost:  cp.Pricing[k].GPU,
+		GPU:      gpuCount,
+	}, nil
+}
+
+func (cp *CustomProvider) DownloadPricingData() error {
+	cp.DownloadPricingDataLock.Lock()
+	defer cp.DownloadPricingDataLock.Unlock()
+
+	if cp.Pricing == nil {
+		m := make(map[string]*NodePrice)
+		cp.Pricing = m
+	}
+	p, err := GetDefaultPricingData("default.json")
+	if err != nil {
+		return err
+	}
+	cp.SpotLabel = p.SpotLabel
+	cp.SpotLabelValue = p.SpotLabelValue
+	cp.GPULabel = p.GpuLabel
+	cp.GPULabelValue = p.GpuLabelValue
+	cp.Pricing["default"] = &NodePrice{
+		CPU: p.CPU,
+		RAM: p.RAM,
+	}
+	cp.Pricing["default,spot"] = &NodePrice{
+		CPU: p.SpotCPU,
+		RAM: p.SpotRAM,
+	}
+	cp.Pricing["default,gpu"] = &NodePrice{
+		CPU: p.CPU,
+		RAM: p.RAM,
+		GPU: p.GPU,
+	}
+	return nil
+}
+
+func (cp *CustomProvider) GetKey(labels map[string]string) Key {
+	return &customProviderKey{
+		SpotLabel:      cp.SpotLabel,
+		SpotLabelValue: cp.SpotLabelValue,
+		GPULabel:       cp.GPULabel,
+		GPULabelValue:  cp.GPULabelValue,
+		Labels:         labels,
+	}
+}
+
+// ExternalAllocations represents tagged assets outside the scope of kubernetes.
+// "start" and "end" are dates of the format YYYY-MM-DD
+// "aggregator" is the tag used to determine how to allocate those assets, ie namespace, pod, etc.
+func (*CustomProvider) ExternalAllocations(start string, end string, aggregator string) ([]*OutOfClusterAllocation, error) {
+	return nil, nil // TODO: transform the QuerySQL lines into the new OutOfClusterAllocation Struct
+}
+
+func (*CustomProvider) QuerySQL(query string) ([]byte, error) {
+	return nil, nil
+}
+
+func (*CustomProvider) PVPricing(pvk PVKey) (*PV, error) {
+	cpricing, err := GetDefaultPricingData("default")
+	if err != nil {
+		return nil, err
+	}
+	return &PV{
+		Cost: cpricing.Storage,
+	}, nil
+}
+
+func (*CustomProvider) NetworkPricing() (*Network, error) {
+	cpricing, err := GetDefaultPricingData("default")
+	if err != nil {
+		return nil, err
+	}
+	znec, err := strconv.ParseFloat(cpricing.ZoneNetworkEgress, 64)
+	if err != nil {
+		return nil, err
+	}
+	rnec, err := strconv.ParseFloat(cpricing.RegionNetworkEgress, 64)
+	if err != nil {
+		return nil, err
+	}
+	inec, err := strconv.ParseFloat(cpricing.InternetNetworkEgress, 64)
+	if err != nil {
+		return nil, err
+	}
+
+	return &Network{
+		ZoneNetworkEgressCost:     znec,
+		RegionNetworkEgressCost:   rnec,
+		InternetNetworkEgressCost: inec,
+	}, nil
+}
+
+func (*CustomProvider) GetPVKey(pv *v1.PersistentVolume, parameters map[string]string) PVKey {
+	return &awsPVKey{
+		Labels:           pv.Labels,
+		StorageClassName: pv.Spec.StorageClassName,
+	}
+}
+
+func (cpk *customProviderKey) GPUType() string {
+	if t, ok := cpk.Labels[cpk.GPULabel]; ok {
+		return t
+	}
+	return ""
+}
+
+func (cpk *customProviderKey) ID() string {
+	return ""
+}
+
+func (cpk *customProviderKey) Features() string {
+	if cpk.Labels[cpk.SpotLabel] != "" && cpk.Labels[cpk.SpotLabel] == cpk.SpotLabelValue {
+		return "default,spot"
+	}
+	return "default" // TODO: multiple custom pricing support.
+}

cloud/gcpprovider.go

@@ -165,7 +165,7 @@ func (gcp *GCP) UpdateConfig(r io.Reader, updateType string) (*CustomPricing, er
 	}
 	remoteEnabled := os.Getenv(remoteEnabled)
 	if remoteEnabled == "true" {
-		err = UpdateClusterMeta(os.Getenv(KC_CLUSTER_ID), c.ClusterName)
+		err = UpdateClusterMeta(os.Getenv(clusterIDKey), c.ClusterName)
 		if err != nil {
 			return nil, err
 		}
@@ -268,7 +268,7 @@ func (gcp *GCP) ClusterInfo() (map[string]string, error) {
 	m := make(map[string]string)
 	m["name"] = attribute
 	m["provider"] = "GCP"
-	m["id"] = os.Getenv(KC_CLUSTER_ID)
+	m["id"] = os.Getenv(clusterIDKey)
 	m["remoteReadEnabled"] = strconv.FormatBool(remoteEnabled)
 	return m, nil
 }

cloud/provider.go

@@ -10,9 +10,7 @@ import (
 	"net/url"
 	"os"
 	"reflect"
-	"strconv"
 	"strings"
-	"sync"
 
 	"k8s.io/klog"
 
@@ -23,10 +21,10 @@ import (
 	"k8s.io/client-go/kubernetes"
 )
 
-const KC_CLUSTER_ID = "CLUSTER_ID"
+const clusterIDKey = "CLUSTER_ID"
+const remoteEnabled = "REMOTE_WRITE_ENABLED"
 const remotePW = "REMOTE_WRITE_PASSWORD"
 const sqlAddress = "SQL_ADDRESS"
-const remoteEnabled = "REMOTE_WRITE_ENABLED"
 
 var createTableStatements = []string{
 	`CREATE TABLE IF NOT EXISTS names (
@@ -57,6 +55,11 @@ type Node struct {
 	GPUCost          string `json:"gpuCost"`
 }
 
+// IsSpot determines whether or not a Node uses spot by usage type
+func (n *Node) IsSpot() bool {
+	return strings.Contains(n.UsageType, "spot") || strings.Contains(n.UsageType, "emptible")
+}
+
 // Network is the interface by which the provider and cost model communicate network egress prices.
 // The provider will best-effort try to fill out this struct.
 type Network struct {
@@ -97,6 +100,45 @@ type OutOfClusterAllocation struct {
 	Cluster     string  `json:"cluster"`
 }
 
+type CustomPricing struct {
+	Provider              string `json:"provider"`
+	Description           string `json:"description"`
+	CPU                   string `json:"CPU"`
+	SpotCPU               string `json:"spotCPU"`
+	RAM                   string `json:"RAM"`
+	SpotRAM               string `json:"spotRAM"`
+	GPU                   string `json:"GPU"`
+	SpotGPU               string `json:"spotGPU"`
+	Storage               string `json:"storage"`
+	ZoneNetworkEgress     string `json:"zoneNetworkEgress"`
+	RegionNetworkEgress   string `json:"regionNetworkEgress"`
+	InternetNetworkEgress string `json:"internetNetworkEgress"`
+	SpotLabel             string `json:"spotLabel,omitempty"`
+	SpotLabelValue        string `json:"spotLabelValue,omitempty"`
+	GpuLabel              string `json:"gpuLabel,omitempty"`
+	GpuLabelValue         string `json:"gpuLabelValue,omitempty"`
+	ServiceKeyName        string `json:"awsServiceKeyName,omitempty"`
+	ServiceKeySecret      string `json:"awsServiceKeySecret,omitempty"`
+	SpotDataRegion        string `json:"awsSpotDataRegion,omitempty"`
+	SpotDataBucket        string `json:"awsSpotDataBucket,omitempty"`
+	SpotDataPrefix        string `json:"awsSpotDataPrefix,omitempty"`
+	ProjectID             string `json:"projectID,omitempty"`
+	AthenaBucketName      string `json:"athenaBucketName"`
+	AthenaRegion          string `json:"athenaRegion"`
+	AthenaDatabase        string `json:"athenaDatabase"`
+	AthenaTable           string `json:"athenaTable"`
+	BillingDataDataset    string `json:"billingDataDataset,omitempty"`
+	CustomPricesEnabled   string `json:"customPricesEnabled"`
+	AzureSubscriptionID   string `json:"azureSubscriptionID"`
+	AzureClientID         string `json:"azureClientID"`
+	AzureClientSecret     string `json:"azureClientSecret"`
+	AzureTenantID         string `json:"azureTenantID"`
+	AzureBillingRegion    string `json:"azureBillingRegion"`
+	CurrencyCode          string `json:"currencyCode"`
+	Discount              string `json:"discount"`
+	ClusterName           string `json:"clusterName"`
+}
+
 // Provider represents a k8s provider.
 type Provider interface {
 	ClusterInfo() (map[string]string, error)
@@ -113,10 +155,36 @@ type Provider interface {
 	GetConfig() (*CustomPricing, error)
 	GetManagementPlatform() (string, error)
 	GetLocalStorageQuery() (string, error)
-
 	ExternalAllocations(string, string, string) ([]*OutOfClusterAllocation, error)
 }
 
+// ClusterName returns the name defined in cluster info, defaulting to the
+// CLUSTER_ID environment variable
+func ClusterName(p Provider) string {
+	info, err := p.ClusterInfo()
+	if err != nil {
+		return os.Getenv(clusterIDKey)
+	}
+
+	name, ok := info["name"]
+	if !ok {
+		return os.Getenv(clusterIDKey)
+	}
+
+	return name
+}
+
+// CustomPricesEnabled returns the boolean equivalent of the cloup provider's custom prices flag,
+// indicating whether or not the cluster is using custom pricing.
+func CustomPricesEnabled(p Provider) bool {
+	config, err := p.GetConfig()
+	if err != nil {
+		return false
+	}
+
+	return config.CustomPricesEnabled == "true"
+}
+
 // GetDefaultPricingData will search for a json file representing pricing data in /models/ and use it for base pricing info.
 func GetDefaultPricingData(fname string) (*CustomPricing, error) {
 	path := os.Getenv("CONFIG_PATH")
@@ -170,47 +238,6 @@ func GetDefaultPricingData(fname string) (*CustomPricing, error) {
 	}
 }
 
-const KeyUpdateType = "athenainfo"
-
-type CustomPricing struct {
-	Provider              string `json:"provider"`
-	Description           string `json:"description"`
-	CPU                   string `json:"CPU"`
-	SpotCPU               string `json:"spotCPU"`
-	RAM                   string `json:"RAM"`
-	SpotRAM               string `json:"spotRAM"`
-	GPU                   string `json:"GPU"`
-	SpotGPU               string `json:"spotGPU"`
-	Storage               string `json:"storage"`
-	ZoneNetworkEgress     string `json:"zoneNetworkEgress"`
-	RegionNetworkEgress   string `json:"regionNetworkEgress"`
-	InternetNetworkEgress string `json:"internetNetworkEgress"`
-	SpotLabel             string `json:"spotLabel,omitempty"`
-	SpotLabelValue        string `json:"spotLabelValue,omitempty"`
-	GpuLabel              string `json:"gpuLabel,omitempty"`
-	GpuLabelValue         string `json:"gpuLabelValue,omitempty"`
-	ServiceKeyName        string `json:"awsServiceKeyName,omitempty"`
-	ServiceKeySecret      string `json:"awsServiceKeySecret,omitempty"`
-	SpotDataRegion        string `json:"awsSpotDataRegion,omitempty"`
-	SpotDataBucket        string `json:"awsSpotDataBucket,omitempty"`
-	SpotDataPrefix        string `json:"awsSpotDataPrefix,omitempty"`
-	ProjectID             string `json:"projectID,omitempty"`
-	AthenaBucketName      string `json:"athenaBucketName"`
-	AthenaRegion          string `json:"athenaRegion"`
-	AthenaDatabase        string `json:"athenaDatabase"`
-	AthenaTable           string `json:"athenaTable"`
-	BillingDataDataset    string `json:"billingDataDataset,omitempty"`
-	CustomPricesEnabled   string `json:"customPricesEnabled"`
-	AzureSubscriptionID   string `json:"azureSubscriptionID"`
-	AzureClientID         string `json:"azureClientID"`
-	AzureClientSecret     string `json:"azureClientSecret"`
-	AzureTenantID         string `json:"azureTenantID"`
-	AzureBillingRegion    string `json:"azureBillingRegion"`
-	CurrencyCode          string `json:"currencyCode"`
-	Discount              string `json:"discount"`
-	ClusterName           string `json:"clusterName"`
-}
-
 func SetCustomPricingField(obj *CustomPricing, name string, value string) error {
 	structValue := reflect.ValueOf(obj).Elem()
 	structFieldValue := structValue.FieldByName(name)
@@ -233,241 +260,6 @@ func SetCustomPricingField(obj *CustomPricing, name string, value string) error
 	return nil
 }
 
-type NodePrice struct {
-	CPU string
-	RAM string
-	GPU string
-}
-
-type CustomProvider struct {
-	Clientset               *kubernetes.Clientset
-	Pricing                 map[string]*NodePrice
-	SpotLabel               string
-	SpotLabelValue          string
-	GPULabel                string
-	GPULabelValue           string
-	DownloadPricingDataLock sync.RWMutex
-}
-
-func (*CustomProvider) GetLocalStorageQuery() (string, error) {
-	return "", nil
-}
-
-func (*CustomProvider) GetConfig() (*CustomPricing, error) {
-	return GetDefaultPricingData("default.json")
-}
-
-func (*CustomProvider) GetManagementPlatform() (string, error) {
-	return "", nil
-}
-
-func (cprov *CustomProvider) UpdateConfig(r io.Reader, updateType string) (*CustomPricing, error) {
-	c, err := GetDefaultPricingData("default.json")
-	if err != nil {
-		return nil, err
-	}
-	path := os.Getenv("CONFIG_PATH")
-	if path == "" {
-		path = "/models/"
-	}
-	a := make(map[string]string)
-	err = json.NewDecoder(r).Decode(&a)
-	if err != nil {
-		return nil, err
-	}
-	for k, v := range a {
-		kUpper := strings.Title(k) // Just so we consistently supply / receive the same values, uppercase the first letter.
-		err := SetCustomPricingField(c, kUpper, v)
-		if err != nil {
-			return nil, err
-		}
-	}
-
-	cj, err := json.Marshal(c)
-	if err != nil {
-		return nil, err
-	}
-
-	configPath := path + "default.json"
-	err = ioutil.WriteFile(configPath, cj, 0644)
-	if err != nil {
-		return nil, err
-	}
-	defer cprov.DownloadPricingData()
-	return c, nil
-
-}
-
-func (c *CustomProvider) ClusterInfo() (map[string]string, error) {
-	conf, err := c.GetConfig()
-	if err != nil {
-		return nil, err
-	}
-	m := make(map[string]string)
-	if conf.ClusterName != "" {
-		m["name"] = conf.ClusterName
-	}
-	m["provider"] = "custom"
-	return m, nil
-}
-
-func (*CustomProvider) AddServiceKey(url.Values) error {
-	return nil
-}
-
-func (*CustomProvider) GetDisks() ([]byte, error) {
-	return nil, nil
-}
-
-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()
-	var gpuCount string
-	if _, ok := c.Pricing[k]; !ok {
-		k = "default"
-	}
-	if key.GPUType() != "" {
-		k += ",gpu"    // TODO: support multiple custom gpu types.
-		gpuCount = "1" // TODO: support more than one gpu.
-	}
-	return &Node{
-		VCPUCost: c.Pricing[k].CPU,
-		RAMCost:  c.Pricing[k].RAM,
-		GPUCost:  c.Pricing[k].GPU,
-		GPU:      gpuCount,
-	}, nil
-}
-
-func (c *CustomProvider) DownloadPricingData() error {
-	c.DownloadPricingDataLock.Lock()
-	defer c.DownloadPricingDataLock.Unlock()
-
-	if c.Pricing == nil {
-		m := make(map[string]*NodePrice)
-		c.Pricing = m
-	}
-	p, err := GetDefaultPricingData("default.json")
-	if err != nil {
-		return err
-	}
-	c.SpotLabel = p.SpotLabel
-	c.SpotLabelValue = p.SpotLabelValue
-	c.GPULabel = p.GpuLabel
-	c.GPULabelValue = p.GpuLabelValue
-	c.Pricing["default"] = &NodePrice{
-		CPU: p.CPU,
-		RAM: p.RAM,
-	}
-	c.Pricing["default,spot"] = &NodePrice{
-		CPU: p.SpotCPU,
-		RAM: p.SpotRAM,
-	}
-	c.Pricing["default,gpu"] = &NodePrice{
-		CPU: p.CPU,
-		RAM: p.RAM,
-		GPU: p.GPU,
-	}
-	return nil
-}
-
-type customProviderKey struct {
-	SpotLabel      string
-	SpotLabelValue string
-	GPULabel       string
-	GPULabelValue  string
-	Labels         map[string]string
-}
-
-func (c *customProviderKey) GPUType() string {
-	if t, ok := c.Labels[c.GPULabel]; ok {
-		return t
-	}
-	return ""
-}
-
-func (c *customProviderKey) ID() string {
-	return ""
-}
-
-func (c *customProviderKey) Features() string {
-	if c.Labels[c.SpotLabel] != "" && c.Labels[c.SpotLabel] == c.SpotLabelValue {
-		return "default,spot"
-	}
-	return "default" // TODO: multiple custom pricing support.
-}
-
-func (c *CustomProvider) GetKey(labels map[string]string) Key {
-	return &customProviderKey{
-		SpotLabel:      c.SpotLabel,
-		SpotLabelValue: c.SpotLabelValue,
-		GPULabel:       c.GPULabel,
-		GPULabelValue:  c.GPULabelValue,
-		Labels:         labels,
-	}
-}
-
-// ExternalAllocations represents tagged assets outside the scope of kubernetes.
-// "start" and "end" are dates of the format YYYY-MM-DD
-// "aggregator" is the tag used to determine how to allocate those assets, ie namespace, pod, etc.
-func (*CustomProvider) ExternalAllocations(start string, end string, aggregator string) ([]*OutOfClusterAllocation, error) {
-	return nil, nil // TODO: transform the QuerySQL lines into the new OutOfClusterAllocation Struct
-}
-
-func (*CustomProvider) QuerySQL(query string) ([]byte, error) {
-	return nil, nil
-}
-
-func (c *CustomProvider) PVPricing(pvk PVKey) (*PV, error) {
-	cpricing, err := GetDefaultPricingData("default")
-	if err != nil {
-		return nil, err
-	}
-	return &PV{
-		Cost: cpricing.Storage,
-	}, nil
-}
-
-func (c *CustomProvider) NetworkPricing() (*Network, error) {
-	cpricing, err := GetDefaultPricingData("default")
-	if err != nil {
-		return nil, err
-	}
-	znec, err := strconv.ParseFloat(cpricing.ZoneNetworkEgress, 64)
-	if err != nil {
-		return nil, err
-	}
-	rnec, err := strconv.ParseFloat(cpricing.RegionNetworkEgress, 64)
-	if err != nil {
-		return nil, err
-	}
-	inec, err := strconv.ParseFloat(cpricing.InternetNetworkEgress, 64)
-	if err != nil {
-		return nil, err
-	}
-
-	return &Network{
-		ZoneNetworkEgressCost:     znec,
-		RegionNetworkEgressCost:   rnec,
-		InternetNetworkEgressCost: inec,
-	}, nil
-}
-
-func (*CustomProvider) GetPVKey(pv *v1.PersistentVolume, parameters map[string]string) PVKey {
-	return &awsPVKey{
-		Labels:           pv.Labels,
-		StorageClassName: pv.Spec.StorageClassName,
-	}
-}
-
 // NewProvider looks at the nodespec or provider metadata server to decide which provider to instantiate.
 func NewProvider(clientset *kubernetes.Clientset, apiKey string) (Provider, error) {
 	if metadata.OnGCE() {
@@ -480,10 +272,12 @@ func NewProvider(clientset *kubernetes.Clientset, apiKey string) (Provider, erro
 			APIKey:    apiKey,
 		}, nil
 	}
+
 	nodes, err := clientset.CoreV1().Nodes().List(metav1.ListOptions{})
 	if err != nil {
 		return nil, err
 	}
+
 	provider := strings.ToLower(nodes.Items[0].Spec.ProviderID)
 	if strings.HasPrefix(provider, "aws") {
 		klog.V(2).Info("Found ProviderID starting with \"aws\", using AWS Provider")
@@ -587,5 +381,4 @@ func GetOrCreateClusterMeta(cluster_id, cluster_name string) (string, string, er
 	}
 
 	return id, name, nil
-
 }

costmodel/aggregations.go

@@ -6,8 +6,9 @@ import (
 	"strconv"
 	"time"
 
-	costAnalyzerCloud "github.com/kubecost/cost-model/cloud"
+	"github.com/kubecost/cost-model/cloud"
 	prometheusClient "github.com/prometheus/client_golang/api"
+	"k8s.io/klog"
 )
 
 type Aggregation struct {
@@ -16,12 +17,12 @@ type Aggregation struct {
 	Environment        string    `json:"environment"`
 	Cluster            string    `json:"cluster"`
 	CPUAllocation      []*Vector `json:"-"`
-	CPUCostVector      []*Vector `json:"-"`
+	CPUCostVector      []*Vector `json:"cpuCostVector,omitempty"`
 	RAMAllocation      []*Vector `json:"-"`
-	RAMCostVector      []*Vector `json:"-"`
-	PVCostVector       []*Vector `json:"-"`
+	RAMCostVector      []*Vector `json:"ramCostVector,omitempty"`
+	PVCostVector       []*Vector `json:"pvCostVector,omitempty"`
 	GPUAllocation      []*Vector `json:"-"`
-	GPUCostVector      []*Vector `json:"-"`
+	GPUCostVector      []*Vector `json:"gpuCostVector,omitempty"`
 	CPUCost            float64   `json:"cpuCost"`
 	RAMCost            float64   `json:"ramCost"`
 	GPUCost            float64   `json:"gpuCost"`
@@ -67,12 +68,12 @@ func NewSharedResourceInfo(shareResources bool, sharedNamespaces []string, label
 	return sr
 }
 
-func ComputeIdleCoefficient(costData map[string]*CostData, cli prometheusClient.Client, cloud costAnalyzerCloud.Provider, discount float64, windowString, offset string) (float64, error) {
+func ComputeIdleCoefficient(costData map[string]*CostData, cli prometheusClient.Client, cp cloud.Provider, discount float64, windowString, offset string) (float64, error) {
 	windowDuration, err := time.ParseDuration(windowString)
 	if err != nil {
 		return 0.0, err
 	}
-	totals, err := ClusterCosts(cli, cloud, windowString, offset)
+	totals, err := ClusterCosts(cli, cp, windowString, offset)
 	if err != nil {
 		return 0.0, err
 	}
@@ -83,7 +84,7 @@ func ComputeIdleCoefficient(costData map[string]*CostData, cli prometheusClient.
 	totalClusterCostOverWindow := (totalClusterCost / 730) * windowDuration.Hours() * (1 - discount)
 	totalContainerCost := 0.0
 	for _, costDatum := range costData {
-		cpuv, ramv, gpuv, pvvs := getPriceVectors(costDatum, discount, 1)
+		cpuv, ramv, gpuv, pvvs := getPriceVectors(cp, costDatum, discount, 1)
 		totalContainerCost += totalVector(cpuv)
 		totalContainerCost += totalVector(ramv)
 		totalContainerCost += totalVector(gpuv)
@@ -95,12 +96,20 @@ func ComputeIdleCoefficient(costData map[string]*CostData, cli prometheusClient.
 	return (totalContainerCost / totalClusterCostOverWindow), nil
 }
 
-func AggregateCostModel(costData map[string]*CostData, discount float64, idleCoefficient float64, sr *SharedResourceInfo, aggregationField string, aggregationSubField string) map[string]*Aggregation {
+// AggregateCostModel reduces the dimensions of raw cost data by field and, optionally, by time. The field parameter determines the field
+// by which to group data, with an optional subfield, e.g. for groupings like field="label" and subfield="app" for grouping by "label.app".
+func AggregateCostModel(cp cloud.Provider, costData map[string]*CostData, field string, subfield string, timeSeries bool, discount float64, idleCoefficient float64, sr *SharedResourceInfo) map[string]*Aggregation {
+	// aggregations collects key-value pairs of resource group-to-aggregated data
+	// e.g. namespace-to-data or label-value-to-data
 	aggregations := make(map[string]*Aggregation)
+
+	// sharedResourceCost is the running total cost of resources that should be reported
+	// as shared across all other resources, rather than reported as a stand-alone category
 	sharedResourceCost := 0.0
+
 	for _, costDatum := range costData {
 		if sr != nil && sr.ShareResources && sr.IsSharedResource(costDatum) {
-			cpuv, ramv, gpuv, pvvs := getPriceVectors(costDatum, discount, idleCoefficient)
+			cpuv, ramv, gpuv, pvvs := getPriceVectors(cp, costDatum, discount, idleCoefficient)
 			sharedResourceCost += totalVector(cpuv)
 			sharedResourceCost += totalVector(ramv)
 			sharedResourceCost += totalVector(gpuv)
@@ -108,27 +117,28 @@ func AggregateCostModel(costData map[string]*CostData, discount float64, idleCoe
 				sharedResourceCost += totalVector(pv)
 			}
 		} else {
-			if aggregationField == "cluster" {
-				aggregationHelper(costDatum, aggregationField, aggregationSubField, costDatum.ClusterID, aggregations, discount, idleCoefficient)
-			} else if aggregationField == "namespace" {
-				aggregationHelper(costDatum, aggregationField, aggregationSubField, costDatum.Namespace, aggregations, discount, idleCoefficient)
-			} else if aggregationField == "service" {
+			if field == "cluster" {
+				aggregateDatum(cp, aggregations, costDatum, field, subfield, costDatum.ClusterID, discount, idleCoefficient)
+			} else if field == "namespace" {
+				aggregateDatum(cp, aggregations, costDatum, field, subfield, costDatum.Namespace, discount, idleCoefficient)
+			} else if field == "service" {
 				if len(costDatum.Services) > 0 {
-					aggregationHelper(costDatum, aggregationField, aggregationSubField, costDatum.Services[0], aggregations, discount, idleCoefficient)
+					aggregateDatum(cp, aggregations, costDatum, field, subfield, costDatum.Services[0], discount, idleCoefficient)
 				}
-			} else if aggregationField == "deployment" {
+			} else if field == "deployment" {
 				if len(costDatum.Deployments) > 0 {
-					aggregationHelper(costDatum, aggregationField, aggregationSubField, costDatum.Deployments[0], aggregations, discount, idleCoefficient)
+					aggregateDatum(cp, aggregations, costDatum, field, subfield, costDatum.Deployments[0], discount, idleCoefficient)
 				}
-			} else if aggregationField == "label" {
+			} else if field == "label" {
 				if costDatum.Labels != nil {
-					if subfieldName, ok := costDatum.Labels[aggregationSubField]; ok {
-						aggregationHelper(costDatum, aggregationField, aggregationSubField, subfieldName, aggregations, discount, idleCoefficient)
+					if subfieldName, ok := costDatum.Labels[subfield]; ok {
+						aggregateDatum(cp, aggregations, costDatum, field, subfield, subfieldName, discount, idleCoefficient)
 					}
 				}
 			}
 		}
 	}
+
 	for _, agg := range aggregations {
 		agg.CPUCost = totalVector(agg.CPUCostVector)
 		agg.RAMCost = totalVector(agg.RAMCostVector)
@@ -136,28 +146,39 @@ func AggregateCostModel(costData map[string]*CostData, discount float64, idleCoe
 		agg.PVCost = totalVector(agg.PVCostVector)
 		agg.SharedCost = sharedResourceCost / float64(len(aggregations))
 		agg.TotalCost = agg.CPUCost + agg.RAMCost + agg.GPUCost + agg.PVCost + agg.SharedCost
+
+		// remove time series data if it is not explicitly requested
+		if !timeSeries {
+			agg.CPUCostVector = nil
+			agg.RAMCostVector = nil
+			agg.PVCostVector = nil
+			agg.GPUCostVector = nil
+		}
 	}
+
 	return aggregations
 }
 
-func aggregationHelper(costDatum *CostData, aggregator string, aggregatorSubField string, key string, aggregations map[string]*Aggregation, discount float64, idleCoefficient float64) {
+func aggregateDatum(cp cloud.Provider, aggregations map[string]*Aggregation, costDatum *CostData, field string, subfield string, key string, discount float64, idleCoefficient float64) {
+	// add new entry to aggregation results if a new
 	if _, ok := aggregations[key]; !ok {
 		agg := &Aggregation{}
-		agg.Aggregator = aggregator
-		agg.AggregatorSubField = aggregatorSubField
+		agg.Aggregator = field
+		agg.AggregatorSubField = subfield
 		agg.Environment = key
 		agg.Cluster = costDatum.ClusterID
 		aggregations[key] = agg
 	}
-	mergeVectors(costDatum, aggregations[key], discount, idleCoefficient)
+
+	mergeVectors(cp, costDatum, aggregations[key], discount, idleCoefficient)
 }
 
-func mergeVectors(costDatum *CostData, aggregation *Aggregation, discount float64, idleCoefficient float64) {
+func mergeVectors(cp cloud.Provider, costDatum *CostData, aggregation *Aggregation, discount float64, idleCoefficient float64) {
 	aggregation.CPUAllocation = addVectors(costDatum.CPUAllocation, aggregation.CPUAllocation)
 	aggregation.RAMAllocation = addVectors(costDatum.RAMAllocation, aggregation.RAMAllocation)
 	aggregation.GPUAllocation = addVectors(costDatum.GPUReq, aggregation.GPUAllocation)
 
-	cpuv, ramv, gpuv, pvvs := getPriceVectors(costDatum, discount, idleCoefficient)
+	cpuv, ramv, gpuv, pvvs := getPriceVectors(cp, costDatum, discount, idleCoefficient)
 	aggregation.CPUCostVector = addVectors(cpuv, aggregation.CPUCostVector)
 	aggregation.RAMCostVector = addVectors(ramv, aggregation.RAMCostVector)
 	aggregation.GPUCostVector = addVectors(gpuv, aggregation.GPUCostVector)
@@ -166,36 +187,71 @@ func mergeVectors(costDatum *CostData, aggregation *Aggregation, discount float6
 	}
 }
 
-func getPriceVectors(costDatum *CostData, discount float64, idleCoefficient float64) ([]*Vector, []*Vector, []*Vector, [][]*Vector) {
+func getPriceVectors(cp cloud.Provider, costDatum *CostData, discount float64, idleCoefficient float64) ([]*Vector, []*Vector, []*Vector, [][]*Vector) {
+	cpuCostStr := costDatum.NodeData.VCPUCost
+	ramCostStr := costDatum.NodeData.RAMCost
+	gpuCostStr := costDatum.NodeData.GPUCost
+	pvCostStr := costDatum.NodeData.StorageCost
+
+	// If custom pricing is enabled and can be retrieved, replace
+	// default cost values with custom values
+	customPricing, err := cp.GetConfig()
+	if err != nil {
+		klog.Errorf("failed to load custom pricing: %s", err)
+	}
+	if cloud.CustomPricesEnabled(cp) && err == nil {
+		if costDatum.NodeData.IsSpot() {
+			cpuCostStr = customPricing.SpotCPU
+			ramCostStr = customPricing.SpotRAM
+			gpuCostStr = customPricing.SpotGPU
+		} else {
+			cpuCostStr = customPricing.CPU
+			ramCostStr = customPricing.RAM
+			gpuCostStr = customPricing.GPU
+		}
+		pvCostStr = customPricing.Storage
+	}
+
+	cpuCost, _ := strconv.ParseFloat(cpuCostStr, 64)
+	ramCost, _ := strconv.ParseFloat(ramCostStr, 64)
+	gpuCost, _ := strconv.ParseFloat(gpuCostStr, 64)
+	pvCost, _ := strconv.ParseFloat(pvCostStr, 64)
+
 	cpuv := make([]*Vector, 0, len(costDatum.CPUAllocation))
 	for _, val := range costDatum.CPUAllocation {
-		cost, _ := strconv.ParseFloat(costDatum.NodeData.VCPUCost, 64)
 		cpuv = append(cpuv, &Vector{
 			Timestamp: math.Round(val.Timestamp/10) * 10,
-			Value:     val.Value * cost * (1 - discount) * 1 / idleCoefficient,
+			Value:     val.Value * cpuCost * (1 - discount) * 1 / idleCoefficient,
 		})
 	}
+
 	ramv := make([]*Vector, 0, len(costDatum.RAMAllocation))
 	for _, val := range costDatum.RAMAllocation {
-		cost, _ := strconv.ParseFloat(costDatum.NodeData.RAMCost, 64)
 		ramv = append(ramv, &Vector{
 			Timestamp: math.Round(val.Timestamp/10) * 10,
-			Value:     (val.Value / 1024 / 1024 / 1024) * cost * (1 - discount) * 1 / idleCoefficient,
+			Value:     (val.Value / 1024 / 1024 / 1024) * ramCost * (1 - discount) * 1 / idleCoefficient,
 		})
 	}
+
 	gpuv := make([]*Vector, 0, len(costDatum.GPUReq))
 	for _, val := range costDatum.GPUReq {
-		cost, _ := strconv.ParseFloat(costDatum.NodeData.GPUCost, 64)
 		gpuv = append(gpuv, &Vector{
 			Timestamp: math.Round(val.Timestamp/10) * 10,
-			Value:     val.Value * cost * (1 - discount) * 1 / idleCoefficient,
+			Value:     val.Value * gpuCost * (1 - discount) * 1 / idleCoefficient,
 		})
 	}
+
 	pvvs := make([][]*Vector, 0, len(costDatum.PVCData))
 	for _, pvcData := range costDatum.PVCData {
 		pvv := make([]*Vector, 0, len(pvcData.Values))
 		if pvcData.Volume != nil {
 			cost, _ := strconv.ParseFloat(pvcData.Volume.Cost, 64)
+
+			// override with custom pricing if enabled
+			if cloud.CustomPricesEnabled(cp) {
+				cost = pvCost
+			}
+
 			for _, val := range pvcData.Values {
 				pvv = append(pvv, &Vector{
 					Timestamp: math.Round(val.Timestamp/10) * 10,
@@ -205,6 +261,7 @@ func getPriceVectors(costDatum *CostData, discount float64, idleCoefficient floa
 			pvvs = append(pvvs, pvv)
 		}
 	}
+
 	return cpuv, ramv, gpuv, pvvs
 }
 
@@ -277,4 +334,4 @@ func addVectors(req []*Vector, used []*Vector) []*Vector {
 	}
 
 	return allocation
-}
+}

costmodel/costmodel.go

@@ -6,13 +6,13 @@ import (
 	"fmt"
 	"math"
 	"net/http"
-	"os"
 	"sort"
 	"strconv"
 	"strings"
 	"sync"
 	"time"
 
+	"github.com/kubecost/cost-model/cloud"
 	costAnalyzerCloud "github.com/kubecost/cost-model/cloud"
 	prometheusClient "github.com/prometheus/client_golang/api"
 	v1 "k8s.io/api/core/v1"
@@ -25,8 +25,7 @@ import (
 const (
 	statusAPIError = 422
 
-	apiPrefix = "/api/v1"
-
+	apiPrefix         = "/api/v1"
 	epAlertManagers   = apiPrefix + "/alertmanagers"
 	epQuery           = apiPrefix + "/query"
 	epQueryRange      = apiPrefix + "/query_range"
@@ -38,8 +37,9 @@ const (
 	epCleanTombstones = apiPrefix + "/admin/tsdb/clean_tombstones"
 	epConfig          = apiPrefix + "/status/config"
 	epFlags           = apiPrefix + "/status/flags"
-	remoteEnabled     = "REMOTE_WRITE_ENABLED"
-	CLUSTER_ID        = "CLUSTER_ID"
+
+	clusterIDKey  = "CLUSTER_ID"
+	remoteEnabled = "REMOTE_WRITE_ENABLED"
 )
 
 type CostModel struct {
@@ -253,7 +253,7 @@ func ComputeUptimes(cli prometheusClient.Client) (map[string]float64, error) {
 	return results, nil
 }
 
-func (cm *CostModel) ComputeCostData(cli prometheusClient.Client, clientset kubernetes.Interface, cloud costAnalyzerCloud.Provider, window string, offset string, filterNamespace string) (map[string]*CostData, error) {
+func (cm *CostModel) ComputeCostData(cli prometheusClient.Client, clientset kubernetes.Interface, cp costAnalyzerCloud.Provider, window string, offset string, filterNamespace string) (map[string]*CostData, error) {
 	queryRAMRequests := fmt.Sprintf(queryRAMRequestsStr, window, offset, window, offset)
 	queryRAMUsage := fmt.Sprintf(queryRAMUsageStr, window, offset, window, offset)
 	queryCPURequests := fmt.Sprintf(queryCPURequestsStr, window, offset, window, offset)
@@ -265,7 +265,8 @@ func (cm *CostModel) ComputeCostData(cli prometheusClient.Client, clientset kube
 	queryNetInternetRequests := fmt.Sprintf(queryInternetNetworkUsage, window, "")
 	normalization := fmt.Sprintf(normalizationStr, window, offset)
 
-	clustID := os.Getenv(CLUSTER_ID)
+	// Retrieve cluster ID from cloud provider's cluster info
+	clusterName := cloud.ClusterName(cp)
 
 	var wg sync.WaitGroup
 	wg.Add(11)
@@ -360,7 +361,7 @@ func (cm *CostModel) ComputeCostData(cli prometheusClient.Client, clientset kube
 		return nil, fmt.Errorf("Error parsing normalization values: " + err.Error())
 	}
 
-	nodes, err := getNodeCost(cm.Cache, cloud)
+	nodes, err := getNodeCost(cm.Cache, cp)
 	if err != nil {
 		klog.V(1).Infof("Warning, no Node cost model available: " + err.Error())
 		return nil, err
@@ -371,7 +372,7 @@ func (cm *CostModel) ComputeCostData(cli prometheusClient.Client, clientset kube
 		klog.Infof("Unable to get PV Data: %s", err.Error())
 	}
 	if pvClaimMapping != nil {
-		err = addPVData(cm.Cache, pvClaimMapping, cloud)
+		err = addPVData(cm.Cache, pvClaimMapping, cp)
 		if err != nil {
 			return nil, err
 		}
@@ -485,7 +486,7 @@ func (cm *CostModel) ComputeCostData(cli prometheusClient.Client, clientset kube
 
 			var podNetCosts []*Vector
 			if usage, ok := networkUsageMap[ns+","+podName]; ok {
-				netCosts, err := GetNetworkCost(usage, cloud)
+				netCosts, err := GetNetworkCost(usage, cp)
 				if err != nil {
 					klog.V(3).Infof("Error pulling network costs: %s", err.Error())
 				} else {
@@ -561,7 +562,7 @@ func (cm *CostModel) ComputeCostData(cli prometheusClient.Client, clientset kube
 					NetworkData:     netReq,
 					Labels:          podLabels,
 					NamespaceLabels: nsLabels,
-					ClusterID:       clustID,
+					ClusterID:       clusterName,
 				}
 				costs.CPUAllocation = getContainerAllocation(costs.CPUReq, costs.CPUUsed)
 				costs.RAMAllocation = getContainerAllocation(costs.RAMReq, costs.RAMUsed)
@@ -630,7 +631,7 @@ func (cm *CostModel) ComputeCostData(cli prometheusClient.Client, clientset kube
 				CPUUsed:         CPUUsedV,
 				GPUReq:          GPUReqV,
 				NamespaceLabels: namespacelabels,
-				ClusterID:       clustID,
+				ClusterID:       clusterName,
 			}
 			costs.CPUAllocation = getContainerAllocation(costs.CPUReq, costs.CPUUsed)
 			costs.RAMAllocation = getContainerAllocation(costs.RAMReq, costs.RAMUsed)
@@ -862,6 +863,7 @@ func getContainerAllocation(req []*Vector, used []*Vector) []*Vector {
 
 	return allocation
 }
+
 func addPVData(cache ClusterCache, pvClaimMapping map[string]*PersistentVolumeClaimData, cloud costAnalyzerCloud.Provider) error {
 	cfg, err := cloud.GetConfig()
 	if err != nil {
@@ -910,13 +912,13 @@ func addPVData(cache ClusterCache, pvClaimMapping map[string]*PersistentVolumeCl
 	return nil
 }
 
-func GetPVCost(pv *costAnalyzerCloud.PV, kpv *v1.PersistentVolume, cloud costAnalyzerCloud.Provider) error {
-	cfg, err := cloud.GetConfig()
+func GetPVCost(pv *costAnalyzerCloud.PV, kpv *v1.PersistentVolume, cp costAnalyzerCloud.Provider) error {
+	cfg, err := cp.GetConfig()
 	if err != nil {
 		return err
 	}
-	key := cloud.GetPVKey(kpv, pv.Parameters)
-	pvWithCost, err := cloud.PVPricing(key)
+	key := cp.GetPVKey(kpv, pv.Parameters)
+	pvWithCost, err := cp.PVPricing(key)
 	if err != nil {
 		pv.Cost = cfg.Storage
 		return err
@@ -929,18 +931,21 @@ func GetPVCost(pv *costAnalyzerCloud.PV, kpv *v1.PersistentVolume, cloud costAna
 	return nil
 }
 
-func getNodeCost(cache ClusterCache, cloud costAnalyzerCloud.Provider) (map[string]*costAnalyzerCloud.Node, error) {
-	cfg, err := cloud.GetConfig()
+func getNodeCost(cache ClusterCache, cp costAnalyzerCloud.Provider) (map[string]*costAnalyzerCloud.Node, error) {
+	cfg, err := cp.GetConfig()
 	if err != nil {
 		return nil, err
 	}
+
 	nodeList := cache.GetAllNodes()
 	nodes := make(map[string]*costAnalyzerCloud.Node)
+
 	for _, n := range nodeList {
 		name := n.GetObjectMeta().GetName()
 		nodeLabels := n.GetObjectMeta().GetLabels()
 		nodeLabels["providerID"] = n.Spec.ProviderID
-		cnode, err := cloud.NodePricing(cloud.GetKey(nodeLabels))
+
+		cnode, err := cp.NodePricing(cp.GetKey(nodeLabels))
 		if err != nil {
 			klog.V(1).Infof("Error getting node. Error: " + err.Error())
 			nodes[name] = cnode
@@ -955,6 +960,7 @@ func getNodeCost(cache ClusterCache, cloud costAnalyzerCloud.Provider) (map[stri
 		} else {
 			cpu, _ = strconv.ParseFloat(newCnode.VCPU, 64)
 		}
+
 		var ram float64
 		if newCnode.RAM == "" {
 			newCnode.RAM = n.Status.Capacity.Memory().String()
@@ -962,28 +968,34 @@ func getNodeCost(cache ClusterCache, cloud costAnalyzerCloud.Provider) (map[stri
 		ram = float64(n.Status.Capacity.Memory().Value())
 		newCnode.RAMBytes = fmt.Sprintf("%f", ram)
 
-		if newCnode.GPU != "" && newCnode.GPUCost == "" { // We couldn't find a gpu cost, so fix cpu and ram, then accordingly
-			klog.V(4).Infof("GPU without cost found for %s, calculating...", cloud.GetKey(nodeLabels).Features())
+		if newCnode.GPU != "" && newCnode.GPUCost == "" {
+			// We couldn't find a gpu cost, so fix cpu and ram, then accordingly
+			klog.V(4).Infof("GPU without cost found for %s, calculating...", cp.GetKey(nodeLabels).Features())
+
 			defaultCPU, err := strconv.ParseFloat(cfg.CPU, 64)
 			if err != nil {
 				klog.V(3).Infof("Could not parse default cpu price")
 				return nil, err
 			}
+
 			defaultRAM, err := strconv.ParseFloat(cfg.RAM, 64)
 			if err != nil {
 				klog.V(3).Infof("Could not parse default ram price")
 				return nil, err
 			}
+
 			defaultGPU, err := strconv.ParseFloat(cfg.RAM, 64)
 			if err != nil {
 				klog.V(3).Infof("Could not parse default gpu price")
 				return nil, err
 			}
+
 			cpuToRAMRatio := defaultCPU / defaultRAM
 			gpuToRAMRatio := defaultGPU / defaultRAM
 
 			ramGB := ram / 1024 / 1024 / 1024
 			ramMultiple := gpuToRAMRatio + cpu*cpuToRAMRatio + ramGB
+
 			var nodePrice float64
 			if newCnode.Cost != "" {
 				nodePrice, err = strconv.ParseFloat(newCnode.Cost, 64)
@@ -1002,57 +1014,59 @@ func getNodeCost(cache ClusterCache, cloud costAnalyzerCloud.Provider) (map[stri
 			ramPrice := (nodePrice / ramMultiple)
 			cpuPrice := ramPrice * cpuToRAMRatio
 			gpuPrice := ramPrice * gpuToRAMRatio
+
 			newCnode.VCPUCost = fmt.Sprintf("%f", cpuPrice)
 			newCnode.RAMCost = fmt.Sprintf("%f", ramPrice)
 			newCnode.RAMBytes = fmt.Sprintf("%f", ram)
 			newCnode.GPUCost = fmt.Sprintf("%f", gpuPrice)
+		} else if newCnode.RAMCost == "" {
+			// We couldn't find a ramcost, so fix cpu and allocate ram accordingly
+			klog.V(4).Infof("No RAM cost found for %s, calculating...", cp.GetKey(nodeLabels).Features())
 
-		} else {
-			if newCnode.RAMCost == "" { // We couldn't find a ramcost, so fix cpu and allocate ram accordingly
-				klog.V(4).Infof("No RAM cost found for %s, calculating...", cloud.GetKey(nodeLabels).Features())
-				defaultCPU, err := strconv.ParseFloat(cfg.CPU, 64)
+			defaultCPU, err := strconv.ParseFloat(cfg.CPU, 64)
+			if err != nil {
+				klog.V(3).Infof("Could not parse default cpu price")
+				return nil, err
+			}
+
+			defaultRAM, err := strconv.ParseFloat(cfg.RAM, 64)
+			if err != nil {
+				klog.V(3).Infof("Could not parse default ram price")
+				return nil, err
+			}
+
+			cpuToRAMRatio := defaultCPU / defaultRAM
+			ramGB := ram / 1024 / 1024 / 1024
+			ramMultiple := cpu*cpuToRAMRatio + ramGB
+
+			var nodePrice float64
+			if newCnode.Cost != "" {
+				nodePrice, err = strconv.ParseFloat(newCnode.Cost, 64)
 				if err != nil {
-					klog.V(3).Infof("Could not parse default cpu price")
+					klog.V(3).Infof("Could not parse total node price")
 					return nil, err
 				}
-				defaultRAM, err := strconv.ParseFloat(cfg.RAM, 64)
+			} else {
+				nodePrice, err = strconv.ParseFloat(newCnode.VCPUCost, 64) // all the price was allocated the the CPU
 				if err != nil {
-					klog.V(3).Infof("Could not parse default ram price")
+					klog.V(3).Infof("Could not parse node vcpu price")
 					return nil, err
 				}
-				cpuToRAMRatio := defaultCPU / defaultRAM
-
-				ramGB := ram / 1024 / 1024 / 1024
-
-				ramMultiple := cpu*cpuToRAMRatio + ramGB
+			}
 
-				var nodePrice float64
-				if newCnode.Cost != "" {
-					nodePrice, err = strconv.ParseFloat(newCnode.Cost, 64)
-					if err != nil {
-						klog.V(3).Infof("Could not parse total node price")
-						return nil, err
-					}
-				} else {
-					nodePrice, err = strconv.ParseFloat(newCnode.VCPUCost, 64) // all the price was allocated the the CPU
-					if err != nil {
-						klog.V(3).Infof("Could not parse node vcpu price")
-						return nil, err
-					}
-				}
+			ramPrice := (nodePrice / ramMultiple)
+			cpuPrice := ramPrice * cpuToRAMRatio
 
-				ramPrice := (nodePrice / ramMultiple)
-				cpuPrice := ramPrice * cpuToRAMRatio
+			newCnode.VCPUCost = fmt.Sprintf("%f", cpuPrice)
+			newCnode.RAMCost = fmt.Sprintf("%f", ramPrice)
+			newCnode.RAMBytes = fmt.Sprintf("%f", ram)
 
-				newCnode.VCPUCost = fmt.Sprintf("%f", cpuPrice)
-				newCnode.RAMCost = fmt.Sprintf("%f", ramPrice)
-				newCnode.RAMBytes = fmt.Sprintf("%f", ram)
-				klog.V(4).Infof("Computed \"%s\" RAM Cost := %v", name, newCnode.RAMCost)
-			}
+			klog.V(4).Infof("Computed \"%s\" RAM Cost := %v", name, newCnode.RAMCost)
 		}
 
 		nodes[name] = &newCnode
 	}
+
 	return nodes, nil
 }
 
@@ -1110,8 +1124,15 @@ func getPodDeployments(cache ClusterCache, podList []*v1.Pod) (map[string]map[st
 	return podDeploymentsMapping, nil
 }
 
-func (cm *CostModel) ComputeCostDataRange(cli prometheusClient.Client, clientset kubernetes.Interface, cloud costAnalyzerCloud.Provider,
-	startString, endString, windowString string, filterNamespace string, remoteEnabled bool) (map[string]*CostData, error) {
+func costDataPassesFilters(costs *CostData, namespace string, cluster string) bool {
+	passesNamespace := namespace == "" || costs.Namespace == namespace
+	passesCluster := cluster == "" || costs.ClusterID == cluster
+
+	return passesNamespace && passesCluster
+}
+
+func (cm *CostModel) ComputeCostDataRange(cli prometheusClient.Client, clientset kubernetes.Interface, cp costAnalyzerCloud.Provider,
+	startString, endString, windowString string, filterNamespace string, filterCluster string, remoteEnabled bool) (map[string]*CostData, error) {
 	queryRAMRequests := fmt.Sprintf(queryRAMRequestsStr, windowString, "", windowString, "")
 	queryRAMUsage := fmt.Sprintf(queryRAMUsageStr, windowString, "", windowString, "")
 	queryCPURequests := fmt.Sprintf(queryCPURequestsStr, windowString, "", windowString, "")
@@ -1140,7 +1161,7 @@ func (cm *CostModel) ComputeCostDataRange(cli prometheusClient.Client, clientset
 		klog.V(1).Infof("Error parsing time " + windowString + ". Error: " + err.Error())
 		return nil, err
 	}
-	clustID := os.Getenv(CLUSTER_ID)
+	clusterName := cloud.ClusterName(cp)
 	if remoteEnabled == true {
 		remoteLayout := "2006-01-02T15:04:05Z"
 		remoteStartStr := start.Format(remoteLayout)
@@ -1242,7 +1263,7 @@ func (cm *CostModel) ComputeCostDataRange(cli prometheusClient.Client, clientset
 		return nil, fmt.Errorf("Error parsing normalization values: " + err.Error())
 	}
 
-	nodes, err := getNodeCost(cm.Cache, cloud)
+	nodes, err := getNodeCost(cm.Cache, cp)
 	if err != nil {
 		klog.V(1).Infof("Warning, no cost model available: " + err.Error())
 		return nil, err
@@ -1254,7 +1275,7 @@ func (cm *CostModel) ComputeCostDataRange(cli prometheusClient.Client, clientset
 		klog.Infof("Unable to get PV Data: %s", err.Error())
 	}
 	if pvClaimMapping != nil {
-		err = addPVData(cm.Cache, pvClaimMapping, cloud)
+		err = addPVData(cm.Cache, pvClaimMapping, cp)
 		if err != nil {
 			return nil, err
 		}
@@ -1356,7 +1377,7 @@ func (cm *CostModel) ComputeCostDataRange(cli prometheusClient.Client, clientset
 
 			var podNetCosts []*Vector
 			if usage, ok := networkUsageMap[ns+","+podName]; ok {
-				netCosts, err := GetNetworkCost(usage, cloud)
+				netCosts, err := GetNetworkCost(usage, cp)
 				if err != nil {
 					klog.V(3).Infof("Error pulling network costs: %s", err.Error())
 				} else {
@@ -1444,13 +1465,12 @@ func (cm *CostModel) ComputeCostDataRange(cli prometheusClient.Client, clientset
 					Labels:          podLabels,
 					NetworkData:     netReq,
 					NamespaceLabels: nsLabels,
-					ClusterID:       clustID,
+					ClusterID:       clusterName,
 				}
 				costs.CPUAllocation = getContainerAllocation(costs.CPUReq, costs.CPUUsed)
 				costs.RAMAllocation = getContainerAllocation(costs.RAMReq, costs.RAMUsed)
-				if filterNamespace == "" {
-					containerNameCost[newKey] = costs
-				} else if costs.Namespace == filterNamespace {
+
+				if costDataPassesFilters(costs, filterNamespace, filterCluster) {
 					containerNameCost[newKey] = costs
 				}
 			}
@@ -1511,14 +1531,12 @@ func (cm *CostModel) ComputeCostDataRange(cli prometheusClient.Client, clientset
 				CPUUsed:         CPUUsedV,
 				GPUReq:          GPUReqV,
 				NamespaceLabels: namespacelabels,
-				ClusterID:       clustID,
+				ClusterID:       clusterName,
 			}
 			costs.CPUAllocation = getContainerAllocation(costs.CPUReq, costs.CPUUsed)
 			costs.RAMAllocation = getContainerAllocation(costs.RAMReq, costs.RAMUsed)
-			if filterNamespace == "" {
-				containerNameCost[key] = costs
-				missingContainers[key] = costs
-			} else if costs.Namespace == filterNamespace {
+
+			if costDataPassesFilters(costs, filterNamespace, filterCluster) {
 				containerNameCost[key] = costs
 				missingContainers[key] = costs
 			}

costmodel/router.go

@@ -70,6 +70,21 @@ type DataEnvelope struct {
 	Message string      `json:"message,omitempty"`
 }
 
+func normalizeTimeParam(param string) (string, error) {
+	// 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
+}
+
 func wrapDataWithMessage(data interface{}, err error, message string) []byte {
 	var resp []byte
 
@@ -181,7 +196,7 @@ func (a *Accesses) CostDataModel(w http.ResponseWriter, r *http.Request, ps http
 			w.Write(wrapData(nil, err))
 		}
 		discount = discount * 0.01
-		agg := AggregateCostModel(data, discount, 1.0, nil, aggregationField, aggregationSubField)
+		agg := AggregateCostModel(a.Cloud, data, aggregationField, aggregationSubField, false, discount, 1.0, nil)
 		w.Write(wrapData(agg, nil))
 	} else {
 		if fields != "" {
@@ -225,6 +240,9 @@ func (a *Accesses) ClusterCostsOverTime(w http.ResponseWriter, r *http.Request,
 	w.Write(wrapData(data, err))
 }
 
+// AggregateCostModel handles HTTP requests to the aggregated cost model API, which can be parametrized
+// by time period using window and offset, aggregation field using field and subfield (in cases like
+// field=label, subfield=app for grouping by label.app), and filtered by namespace.
 func (a *Accesses) AggregateCostModel(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
 	w.Header().Set("Content-Type", "application/json")
 	w.Header().Set("Access-Control-Allow-Origin", "*")
@@ -232,19 +250,38 @@ func (a *Accesses) AggregateCostModel(w http.ResponseWriter, r *http.Request, ps
 	window := r.URL.Query().Get("window")
 	offset := r.URL.Query().Get("offset")
 	namespace := r.URL.Query().Get("namespace")
-	aggregationField := r.URL.Query().Get("aggregation")
-	aggregationSubField := r.URL.Query().Get("aggregationSubfield")
+	cluster := r.URL.Query().Get("cluster")
+	field := r.URL.Query().Get("aggregation")
+	subfield := r.URL.Query().Get("aggregationSubfield")
 	allocateIdle := r.URL.Query().Get("allocateIdle")
 	sharedNamespaces := r.URL.Query().Get("sharedNamespaces")
 	sharedLabelNames := r.URL.Query().Get("sharedLabelNames")
 	sharedLabelValues := r.URL.Query().Get("sharedLabelValues")
+	remote := r.URL.Query().Get("remote")
 
+	// timeSeries == true maintains the time series dimension of the data,
+	// which by default gets summed over the entire interval
+	timeSeries := r.URL.Query().Get("timeSeries") == "true"
+
+	// 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"
 
-	if aggregationField == "" {
+	// aggregation field is required
+	if field == "" {
+		w.WriteHeader(http.StatusBadRequest)
+		w.Write(wrapData(nil, fmt.Errorf("Missing aggregation field parameter")))
+		return
+	}
+
+	// aggregation subfield is required when aggregation field is "label"
+	if field == "label" && subfield == "" {
 		w.WriteHeader(http.StatusBadRequest)
-		w.Write(wrapData(nil, fmt.Errorf("Missing aggregation parameter")))
+		w.Write(wrapData(nil, fmt.Errorf("Missing aggregation subfield parameter for aggregation by label")))
 		return
 	}
 
@@ -261,17 +298,12 @@ func (a *Accesses) AggregateCostModel(w http.ResponseWriter, r *http.Request, ps
 		endTime = endTime.Add(-1 * o)
 	}
 
-	// if window is defined in terms of days, convert it to hours
+	// if window is defined in terms of days, convert to hours
 	// e.g. convert "2d" to "48h"
-	if window[len(window)-1:] == "d" {
-		count := window[:len(window)-1]
-		val, err := strconv.ParseInt(count, 10, 64)
-		if err != nil {
-			w.Write(wrapData(nil, err))
-			return
-		}
-		val = val * 24
-		window = fmt.Sprintf("%dh", val)
+	window, err := normalizeTimeParam(window)
+	if err != nil {
+		w.Write(wrapData(nil, err))
+		return
 	}
 
 	// convert time window into start and end times, formatted
@@ -293,18 +325,14 @@ func (a *Accesses) AggregateCostModel(w http.ResponseWriter, r *http.Request, ps
 		a.Cache.Flush()
 	}
 
-	aggKey := fmt.Sprintf("aggregate:%s:%s:%s:%s:%s", window, offset, namespace, aggregationField, aggregationSubField)
+	aggKey := fmt.Sprintf("aggregate:%s:%s:%s:%s:%s:%s:%t", window, offset, namespace, cluster, field, subfield, timeSeries)
 
 	// check the cache for aggregated response; if cache is hit and not disabled, return response
 	if result, found := a.Cache.Get(aggKey); found && !disableCache {
-		// TODO send http.StatusNotModified when testing is complete
-		w.WriteHeader(http.StatusOK)
 		w.Write(wrapDataWithMessage(result, nil, fmt.Sprintf("cache hit: %s", aggKey)))
 		return
 	}
 
-	remote := r.URL.Query().Get("remote")
-
 	remoteAvailable := os.Getenv(remoteEnabled)
 	remoteEnabled := false
 	if remoteAvailable == "true" && remote != "false" {
@@ -312,7 +340,7 @@ func (a *Accesses) AggregateCostModel(w http.ResponseWriter, r *http.Request, ps
 	}
 	klog.Infof("REMOTE ENABLED: %t", remoteEnabled)
 
-	data, err := a.Model.ComputeCostDataRange(a.PrometheusClient, a.KubeClientSet, a.Cloud, start, end, "1h", namespace, remoteEnabled)
+	data, err := a.Model.ComputeCostDataRange(a.PrometheusClient, a.KubeClientSet, a.Cloud, start, end, "1h", namespace, cluster, remoteEnabled)
 	if err != nil {
 		w.Write(wrapData(nil, err))
 		return
@@ -352,13 +380,13 @@ func (a *Accesses) AggregateCostModel(w http.ResponseWriter, r *http.Request, ps
 			return
 		}
 	}
-	var s *SharedResourceInfo
+	var sr *SharedResourceInfo
 	if len(sn) > 0 || len(sln) > 0 {
-		s = NewSharedResourceInfo(true, sn, sln, slv)
+		sr = NewSharedResourceInfo(true, sn, sln, slv)
 	}
 
 	// aggregate cost model data by given fields and cache the result for the default expiration
-	result := AggregateCostModel(data, discount, idleCoefficient, s, aggregationField, aggregationSubField)
+	result := AggregateCostModel(a.Cloud, data, field, subfield, timeSeries, discount, idleCoefficient, sr)
 	a.Cache.Set(aggKey, result, cache.DefaultExpiration)
 
 	w.Write(wrapDataWithMessage(result, nil, fmt.Sprintf("cache miss: %s", aggKey)))
@@ -373,6 +401,7 @@ func (a *Accesses) CostDataModelRange(w http.ResponseWriter, r *http.Request, ps
 	window := r.URL.Query().Get("window")
 	fields := r.URL.Query().Get("filterFields")
 	namespace := r.URL.Query().Get("namespace")
+	cluster := r.URL.Query().Get("cluster")
 	aggregationField := r.URL.Query().Get("aggregation")
 	aggregationSubField := r.URL.Query().Get("aggregationSubfield")
 	remote := r.URL.Query().Get("remote")
@@ -382,7 +411,7 @@ func (a *Accesses) CostDataModelRange(w http.ResponseWriter, r *http.Request, ps
 	if remoteAvailable == "true" && remote != "false" {
 		remoteEnabled = true
 	}
-	data, err := a.Model.ComputeCostDataRange(a.PrometheusClient, a.KubeClientSet, a.Cloud, start, end, window, namespace, remoteEnabled)
+	data, err := a.Model.ComputeCostDataRange(a.PrometheusClient, a.KubeClientSet, a.Cloud, start, end, window, namespace, cluster, remoteEnabled)
 	if err != nil {
 		w.Write(wrapData(nil, err))
 	}
@@ -396,7 +425,7 @@ func (a *Accesses) CostDataModelRange(w http.ResponseWriter, r *http.Request, ps
 			w.Write(wrapData(nil, err))
 		}
 		discount = discount * 0.01
-		agg := AggregateCostModel(data, discount, 1.0, nil, aggregationField, aggregationSubField)
+		agg := AggregateCostModel(a.Cloud, data, aggregationField, aggregationSubField, false, discount, 1.0, nil)
 		w.Write(wrapData(agg, nil))
 	} else {
 		if fields != "" {

main.go

@@ -1,15 +1,16 @@
 package main
 
 import (
-	costModel "github.com/kubecost/cost-model/costmodel"
-	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"net/http"
+
+	"github.com/kubecost/cost-model/costmodel"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"k8s.io/klog"
 )
 
 func main() {
 	rootMux := http.NewServeMux()
-	rootMux.Handle("/", costModel.Router)
+	rootMux.Handle("/", costmodel.Router)
 	rootMux.Handle("/metrics", promhttp.Handler())
 	klog.Fatal(http.ListenAndServe(":9003", rootMux))
-}
+}

test/aggregation_test.go

@@ -101,7 +101,7 @@ func TestAggregation(t *testing.T) {
 	costData := make(map[string]*costModel.CostData)
 	costData["test1,foo,nginx,testnode"] = cd1
 	costData["test1,bar,nginx,testnode"] = cd2
-	agg := costModel.AggregateCostModel(costData, 0.0, 1.0, nil, "namespace", "")
+	agg := costModel.AggregateCostModel(costData, "namespace", "", false, 0.0, 1.0, nil)
 	log.Printf("agg: %+v", agg["test1"])
 	assert.Equal(t, agg["test1"].TotalCost, 8.0)
 }

test/historical_pod_test.go

@@ -190,11 +190,11 @@ func TestPodUpDown(t *testing.T) {
 	log.Printf("Starting at %s \n", startStr)
 	log.Printf("Ending at %s \n", endStr)
 	provider.DownloadPricingData()
-	data, err := cm.ComputeCostDataRange(promCli, rclient, provider, startStr, endStr, "1m", "", false)
+	data, err := cm.ComputeCostDataRange(promCli, rclient, provider, startStr, endStr, "1m", "", "", false)
 	if err != nil {
 		panic(err)
 	}
-	agg := costModel.AggregateCostModel(data, 0.0, 1.0, nil, "namespace", "")
+	agg := costModel.AggregateCostModel(data, "namespace", "", false, 0.0, 1.0, nil)
 	_, ok := agg["test"]
 	assert.Assert(t, ok)
 
@@ -202,11 +202,11 @@ func TestPodUpDown(t *testing.T) {
 	if err != nil {
 		panic(err)
 	}
-	agg2 := costModel.AggregateCostModel(data2, 0.0, 1.0, nil, "namespace", "")
+	agg2 := costModel.AggregateCostModel(data2, "namespace", "", false, 0.0, 1.0, nil)
 	_, ok2 := agg2["test"]
 	assert.Assert(t, ok2)
 
-	agg3 := costModel.AggregateCostModel(data, 0.0, 1.0, nil, "label", "testaggregation")
+	agg3 := costModel.AggregateCostModel(data, "label", "testaggregation", false, 0.0, 1.0, nil)
 	_, ok3 := agg3["foo"]
 	assert.Assert(t, ok3)
 }

test/remote_cluster_test.go

@@ -55,7 +55,7 @@ func TestClusterConvergence(t *testing.T) {
 	log.Printf("Ending at %s \n", endStr)
 	provider.DownloadPricingData()
 
-	data, err := cm.ComputeCostDataRange(promCli, rclient, provider, startStr, endStr, "1h", "", false)
+	data, err := cm.ComputeCostDataRange(promCli, rclient, provider, startStr, endStr, "1h", "", "", false)
 	if err != nil {
 		panic(err)
 	}
@@ -63,13 +63,13 @@ func TestClusterConvergence(t *testing.T) {
 	os.Setenv("SQL_ADDRESS", "ab5cfc235d64e11e9b8280265f54018f-778641917.us-east-2.elb.amazonaws.com")
 	os.Setenv("REMOTE_WRITE_PASSWORD", "savemoney123")
 
-	data2, err := cm.ComputeCostDataRange(promCli, rclient, provider, startStr, endStr, "1h", "", true)
+	data2, err := cm.ComputeCostDataRange(promCli, rclient, provider, startStr, endStr, "1h", "", "", true)
 	if err != nil {
 		panic(err)
 	}
 
-	agg := costModel.AggregateCostModel(data, 0.0, 1.0, nil, "namespace", "")
-	agg2 := costModel.AggregateCostModel(data2, 0.0, 1.0, nil, "namespace", "")
+	agg := costModel.AggregateCostModel(data, "namespace", "", false, 0.0, 1.0, nil)
+	agg2 := costModel.AggregateCostModel(data2, "namespace", "", false, 0.0, 1.0, nil)
 
 	assert.Equal(t, agg["kubecost"].TotalCost, agg2["kubecost"].TotalCost)