Merge pull request #904 from kubecost/master

Merge Master into develop

pkg/costmodel/costmodel.go

@@ -233,7 +233,6 @@ const (
 func (cm *CostModel) ComputeCostData(cli prometheusClient.Client, cp costAnalyzerCloud.Provider, window string, offset string, filterNamespace string) (map[string]*CostData, error) {
 	queryRAMUsage := fmt.Sprintf(queryRAMUsageStr, window, offset, window, offset, env.GetPromClusterLabel())
 	queryCPUUsage := fmt.Sprintf(queryCPUUsageStr, window, offset, env.GetPromClusterLabel())
-	queryPVRequests := fmt.Sprintf(queryPVRequestsStr, env.GetPromClusterLabel(), env.GetPromClusterLabel(), env.GetPromClusterLabel(), env.GetPromClusterLabel())
 	queryNetZoneRequests := fmt.Sprintf(queryZoneNetworkUsage, window, "", env.GetPromClusterLabel())
 	queryNetRegionRequests := fmt.Sprintf(queryRegionNetworkUsage, window, "", env.GetPromClusterLabel())
 	queryNetInternetRequests := fmt.Sprintf(queryInternetNetworkUsage, window, "", env.GetPromClusterLabel())
@@ -246,7 +245,6 @@ func (cm *CostModel) ComputeCostData(cli prometheusClient.Client, cp costAnalyze
 	ctx := prom.NewContext(cli)
 	resChRAMUsage := ctx.Query(queryRAMUsage)
 	resChCPUUsage := ctx.Query(queryCPUUsage)
-	resChPVRequests := ctx.Query(queryPVRequests)
 	resChNetZoneRequests := ctx.Query(queryNetZoneRequests)
 	resChNetRegionRequests := ctx.Query(queryNetRegionRequests)
 	resChNetInternetRequests := ctx.Query(queryNetInternetRequests)
@@ -278,7 +276,6 @@ func (cm *CostModel) ComputeCostData(cli prometheusClient.Client, cp costAnalyze
 	// Process Prometheus query results. Handle errors using ctx.Errors.
 	resRAMUsage, _ := resChRAMUsage.Await()
 	resCPUUsage, _ := resChCPUUsage.Await()
-	resPVRequests, _ := resChPVRequests.Await()
 	resNetZoneRequests, _ := resChNetZoneRequests.Await()
 	resNetRegionRequests, _ := resChNetRegionRequests.Await()
 	resNetInternetRequests, _ := resChNetInternetRequests.Await()
@@ -327,7 +324,7 @@ func (cm *CostModel) ComputeCostData(cli prometheusClient.Client, cp costAnalyze
 
 	// Unmounted PVs represent the PVs that are not mounted or tied to a volume on a container
 	unmountedPVs := make(map[string][]*PersistentVolumeClaimData)
-	pvClaimMapping, err := GetPVInfo(resPVRequests, clusterID)
+	pvClaimMapping, err := GetPVInfoLocal(cm.Cache, clusterID)
 	if err != nil {
 		log.Warningf("GetPVInfo: unable to get PV data: %s", err.Error())
 	}

pkg/costmodel/metrics.go

@@ -407,6 +407,9 @@ func (cmme *CostModelMetricsEmitter) Start() bool {
 
 			// TODO: Pass CloudProvider into CostModel on instantiation so this isn't so awkward
 			nodes, err := cmme.Model.GetNodeCost(cmme.CloudProvider)
+			if err != nil {
+				log.Warningf("Metric emission: error getting Node cost: %s", err)
+			}
 			for nodeName, node := range nodes {
 				// Emit costs, guarding against NaN inputs for custom pricing.
 				cpuCost, _ := strconv.ParseFloat(node.VCPUCost, 64)
@@ -463,6 +466,9 @@ func (cmme *CostModelMetricsEmitter) Start() bool {
 
 			// TODO: Pass CloudProvider into CostModel on instantiation so this isn't so awkward
 			loadBalancers, err := cmme.Model.GetLBCost(cmme.CloudProvider)
+			if err != nil {
+				log.Warningf("Metric emission: error getting LoadBalancer cost: %s", err)
+			}
 			for lbKey, lb := range loadBalancers {
 				// TODO: parse (if necessary) and calculate cost associated with loadBalancer based on dynamic cloud prices fetched into each lb struct on GetLBCost() call
 				keyParts := getLabelStringsFromKey(lbKey)
@@ -474,7 +480,7 @@ func (cmme *CostModelMetricsEmitter) Start() bool {
 				}
 				cmme.LBCostRecorder.WithLabelValues(ingressIP, namespace, serviceName).Set(lb.Cost)
 
-				labelKey := getKeyFromLabelStrings(namespace, serviceName)
+				labelKey := getKeyFromLabelStrings(ingressIP, namespace, serviceName)
 				loadBalancerSeen[labelKey] = true
 			}
 
@@ -607,7 +613,11 @@ func (cmme *CostModelMetricsEmitter) Start() bool {
 			for labelString, seen := range loadBalancerSeen {
 				if !seen {
 					labels := getLabelStringsFromKey(labelString)
-					cmme.LBCostRecorder.DeleteLabelValues(labels...)
+					ok := cmme.LBCostRecorder.DeleteLabelValues(labels...)
+					if !ok {
+						log.Warningf("Metric emission: failed to delete LoadBalancer with labels: %v", labels)
+					}
+					delete(loadBalancerSeen, labelString)
 				} else {
 					loadBalancerSeen[labelString] = false
 				}

pkg/costmodel/promparsers.go

@@ -3,8 +3,10 @@ package costmodel
 import (
 	"errors"
 	"fmt"
+	"time"
 
 	costAnalyzerCloud "github.com/kubecost/cost-model/pkg/cloud"
+	"github.com/kubecost/cost-model/pkg/clustercache"
 	"github.com/kubecost/cost-model/pkg/env"
 	"github.com/kubecost/cost-model/pkg/log"
 	"github.com/kubecost/cost-model/pkg/prom"
@@ -33,6 +35,34 @@ func GetKubecostJobName() string {
 	return DEFAULT_KUBECOST_JOB_NAME // TODO: look this up from a prometheus variable?
 }
 
+func GetPVInfoLocal(cache clustercache.ClusterCache, defaultClusterID string) (map[string]*PersistentVolumeClaimData, error) {
+	toReturn := make(map[string]*PersistentVolumeClaimData)
+
+	pvcs := cache.GetAllPersistentVolumeClaims()
+	for _, pvc := range pvcs {
+		var vals []*util.Vector
+		vals = append(vals, &util.Vector{
+			Timestamp: float64(time.Now().Unix()),
+			Value:     float64(pvc.Spec.Resources.Requests.Storage().Value()),
+		})
+		ns := pvc.Namespace
+		pvcName := pvc.Name
+		volumeName := pvc.Spec.VolumeName
+		pvClass := *pvc.Spec.StorageClassName
+		clusterID := defaultClusterID
+		key := fmt.Sprintf("%s,%s,%s", ns, pvcName, clusterID)
+		toReturn[key] = &PersistentVolumeClaimData{
+			Class:      pvClass,
+			Claim:      pvcName,
+			Namespace:  ns,
+			ClusterID:  clusterID,
+			VolumeName: volumeName,
+			Values:     vals,
+		}
+	}
+	return toReturn, nil
+}
+
 // TODO niko/prom move parsing functions from costmodel.go
 
 func GetPVInfo(qrs []*prom.QueryResult, defaultClusterID string) (map[string]*PersistentVolumeClaimData, error) {