Merge pull request #210 from kubecost/AjayTripathy-add-localstorage

Ajay tripathy add localstorage

cloud/awsprovider.go

@@ -204,7 +204,7 @@ var regionToBillingRegionCode = map[string]string{
 	"us-gov-west-1":  "UGW1",
 }
 
-func (aws *AWS) GetLocalStorageQuery() (string, error) {
+func (aws *AWS) GetLocalStorageQuery(offset string) (string, error) {
 	return "", nil
 }
 

cloud/azureprovider.go

@@ -582,6 +582,6 @@ func (az *Azure) PVPricing(PVKey) (*PV, error) {
 	return nil, nil
 }
 
-func (az *Azure) GetLocalStorageQuery() (string, error) {
+func (az *Azure) GetLocalStorageQuery(offset string) (string, error) {
 	return "", nil
 }

cloud/customprovider.go

@@ -38,7 +38,7 @@ type customProviderKey struct {
 	Labels         map[string]string
 }
 
-func (*CustomProvider) GetLocalStorageQuery() (string, error) {
+func (*CustomProvider) GetLocalStorageQuery(offset string) (string, error) {
 	return "", nil
 }
 

cloud/gcpprovider.go

@@ -79,9 +79,9 @@ func gcpAllocationToOutOfClusterAllocation(gcpAlloc gcpAllocation) *OutOfCluster
 	}
 }
 
-func (gcp *GCP) GetLocalStorageQuery() (string, error) {
+func (gcp *GCP) GetLocalStorageQuery(offset string) (string, error) {
 	localStorageCost := 0.04 // TODO: Set to the price for the appropriate storage class. It's not trivial to determine the local storage disk type
-	return fmt.Sprintf(`sum(sum(container_fs_limit_bytes{device!="tmpfs", id="/"}) by (instance) / 1024 / 1024 / 1024) * %f`, localStorageCost), nil
+	return fmt.Sprintf(`sum(sum(container_fs_limit_bytes{device!="tmpfs", id="/"} %s) by (instance, cluster_id)) by (cluster_id) / 1024 / 1024 / 1024 * %f`, offset, localStorageCost), nil
 }
 
 func (gcp *GCP) GetConfig() (*CustomPricing, error) {

cloud/provider.go

@@ -154,7 +154,7 @@ type Provider interface {
 	UpdateConfig(r io.Reader, updateType string) (*CustomPricing, error)
 	GetConfig() (*CustomPricing, error)
 	GetManagementPlatform() (string, error)
-	GetLocalStorageQuery() (string, error)
+	GetLocalStorageQuery(offset string) (string, error)
 	ExternalAllocations(string, string, string) ([]*OutOfClusterAllocation, error)
 }
 

costmodel/cluster.go

@@ -148,9 +148,17 @@ func ClusterCostsForAllClusters(cli prometheusClient.Client, cloud costAnalyzerC
 		offset = fmt.Sprintf("offset %s", offset)
 	}
 
+	localStorageQuery, err := cloud.GetLocalStorageQuery(offset)
+	if err != nil {
+		return nil, err
+	}
+	if localStorageQuery != "" {
+		localStorageQuery = fmt.Sprintf("+ %s", localStorageQuery)
+	}
+
 	qCores := fmt.Sprintf(queryClusterCores, windowString, offset, windowString, offset, windowString, offset)
 	qRAM := fmt.Sprintf(queryClusterRAM, windowString, offset, windowString, offset)
-	qStorage := fmt.Sprintf(queryStorage, windowString, offset, windowString, offset, "")
+	qStorage := fmt.Sprintf(queryStorage, windowString, offset, windowString, offset, localStorageQuery)
 
 	klog.V(4).Infof("Running query %s", qCores)
 	resultClusterCores, err := Query(cli, qCores)
@@ -209,7 +217,12 @@ func ClusterCostsForAllClusters(cli prometheusClient.Client, cloud costAnalyzerC
 // ClusterCosts gives the current full cluster costs averaged over a window of time.
 func ClusterCosts(cli prometheusClient.Client, cloud costAnalyzerCloud.Provider, windowString, offset string) (*Totals, error) {
 
-	localStorageQuery, err := cloud.GetLocalStorageQuery()
+	// turn offsets of the format "[0-9+]h" into the format "offset [0-9+]h" for use in query templatess
+	if offset != "" {
+		offset = fmt.Sprintf("offset %s", offset)
+	}
+
+	localStorageQuery, err := cloud.GetLocalStorageQuery(offset)
 	if err != nil {
 		return nil, err
 	}
@@ -217,11 +230,6 @@ func ClusterCosts(cli prometheusClient.Client, cloud costAnalyzerCloud.Provider,
 		localStorageQuery = fmt.Sprintf("+ %s", localStorageQuery)
 	}
 
-	// turn offsets of the format "[0-9+]h" into the format "offset [0-9+]h" for use in query templatess
-	if offset != "" {
-		offset = fmt.Sprintf("offset %s", offset)
-	}
-
 	qCores := fmt.Sprintf(queryClusterCores, windowString, offset, windowString, offset, windowString, offset)
 	qRAM := fmt.Sprintf(queryClusterRAM, windowString, offset, windowString, offset)
 	qStorage := fmt.Sprintf(queryStorage, windowString, offset, windowString, offset, localStorageQuery)
@@ -279,7 +287,7 @@ func ClusterCosts(cli prometheusClient.Client, cloud costAnalyzerCloud.Provider,
 // ClusterCostsOverTime gives the full cluster costs over time
 func ClusterCostsOverTime(cli prometheusClient.Client, cloud costAnalyzerCloud.Provider, startString, endString, windowString, offset string) (*Totals, error) {
 
-	localStorageQuery, err := cloud.GetLocalStorageQuery()
+	localStorageQuery, err := cloud.GetLocalStorageQuery(offset)
 	if err != nil {
 		return nil, err
 	}

costmodel/costmodel.go

@@ -150,7 +150,7 @@ const (
 						* 
 						on (persistentvolumeclaim, namespace, cluster_id) group_right(storageclass, volumename) 
 				sum(kube_persistentvolumeclaim_resource_requests_storage_bytes) by (persistentvolumeclaim, namespace, cluster_id)`
-	queryPVCAllocation        = `avg_over_time(pod_pvc_allocation[24h])`
+	queryPVCAllocation        = `pod_pvc_allocation`
 	queryPVHourlyCost         = `avg_over_time(pv_hourly_cost[24h])`
 	queryNSLabels             = `avg_over_time(kube_namespace_labels[24h])`
 	queryZoneNetworkUsage     = `sum(increase(kubecost_pod_network_egress_bytes_total{internet="false", sameZone="false", sameRegion="true"}[%s] %s)) by (namespace,pod_name,cluster_id) / 1024 / 1024 / 1024`
@@ -1234,12 +1234,12 @@ func (cm *CostModel) ComputeCostDataRange(cli prometheusClient.Client, clientset
 	}()
 	var pvPodAllocationResults interface{}
 	go func() {
-		pvPodAllocationResults, promErr = Query(cli, queryPVCAllocation)
+		pvPodAllocationResults, promErr = QueryRange(cli, queryPVCAllocation, start, end, window)
 		defer wg.Done()
 	}()
 	var pvCostResults interface{}
 	go func() {
-		pvCostResults, promErr = Query(cli, queryPVHourlyCost)
+		pvCostResults, promErr = QueryRange(cli, queryPVHourlyCost, start, end, window)
 		defer wg.Done()
 	}()
 	var nsLabelsResults interface{}
@@ -1670,6 +1670,7 @@ func getPVAllocationMetrics(queryResult interface{}, defaultClusterID string) (m
 		return toReturn, fmt.Errorf(e)
 	}
 
+	vectors := []*Vector{}
 	for _, val := range data.(map[string]interface{})["result"].([]interface{}) {
 		metricInterface, ok := val.(map[string]interface{})["metric"]
 		if !ok {
@@ -1704,23 +1705,19 @@ func getPVAllocationMetrics(queryResult interface{}, defaultClusterID string) (m
 		if err != nil {
 			return toReturn, err
 		}
-
-		dataPoint, ok := val.(map[string]interface{})["value"]
-		if !ok {
-			return nil, fmt.Errorf("Value field does not exist in data result vector")
-		}
-		value, ok := dataPoint.([]interface{})
-		if !ok || len(value) != 2 {
-			return nil, fmt.Errorf("Improperly formatted datapoint from Prometheus")
+		values, ok := val.(map[string]interface{})["values"].([]interface{})
+		for _, d := range values {
+			dataPoint := d.([]interface{})
+			if !ok || len(dataPoint) != 2 {
+				return nil, fmt.Errorf("Improperly formatted datapoint from Prometheus")
+			}
+			strVal := dataPoint[1].(string)
+			v, _ := strconv.ParseFloat(strVal, 64)
+			vectors = append(vectors, &Vector{
+				Timestamp: math.Round(dataPoint[0].(float64)/10) * 10,
+				Value:     v,
+			})
 		}
-		var vectors []*Vector
-		strVal := value[1].(string)
-		v, _ := strconv.ParseFloat(strVal, 64)
-
-		vectors = append(vectors, &Vector{
-			Timestamp: value[0].(float64),
-			Value:     v,
-		})
 
 		key := fmt.Sprintf("%s,%s,%s", ns, pod, clusterID)
 		pvcData := &PersistentVolumeClaimData{

costmodel/router.go

@@ -409,7 +409,7 @@ func (a *Accesses) AggregateCostModel(w http.ResponseWriter, r *http.Request, ps
 		remoteEnabled = true
 	}
 
-	// Use Thanos Client if it exists (enabled) and remote flag set
+	// Use Thanos Client if it exists (enabled) and remote flag not set
 	var pClient prometheusClient.Client
 	if remote != "false" && a.ThanosClient != nil {
 		pClient = a.ThanosClient