Merge pull request #576 from kubecost/niko/etl

Improved prom errors

pkg/costmodel/costmodel.go

@@ -1639,10 +1639,7 @@ func (cm *CostModel) costDataRange(cli prometheusClient.Client, clientset kubern
 	normalizationValue, err := getNormalizations(resNormalization)
 	if err != nil {
 		msg := fmt.Sprintf("error computing normalization for start=%s, end=%s, window=%s, res=%f", start, end, window, resolutionHours*60*60)
-		if pce, ok := err.(prom.CommError); ok {
-			return nil, pce.Wrap(msg)
-		}
-		return nil, fmt.Errorf("%s: %s", msg, err)
+		return nil, prom.WrapError(err, msg)
 	}
 
 	measureTime(profileStart, profileThreshold, fmt.Sprintf("costDataRange(%fh): compute normalizations", durHrs))
@@ -1758,10 +1755,7 @@ func (cm *CostModel) costDataRange(cli prometheusClient.Client, clientset kubern
 
 	RAMReqMap, err := GetNormalizedContainerMetricVectors(resRAMRequests, normalizationValue, clusterID)
 	if err != nil {
-		if pce, ok := err.(prom.CommError); ok {
-			return nil, pce.Wrap("GetNormalizedContainerMetricVectors(RAMRequests)")
-		}
-		return nil, fmt.Errorf("GetNormalizedContainerMetricVectors(RAMRequests): %s", err)
+		return nil, prom.WrapError(err, "GetNormalizedContainerMetricVectors(RAMRequests)")
 	}
 	for key := range RAMReqMap {
 		containers[key] = true
@@ -1769,10 +1763,7 @@ func (cm *CostModel) costDataRange(cli prometheusClient.Client, clientset kubern
 
 	RAMUsedMap, err := GetNormalizedContainerMetricVectors(resRAMUsage, normalizationValue, clusterID)
 	if err != nil {
-		if pce, ok := err.(prom.CommError); ok {
-			return nil, pce.Wrap("GetNormalizedContainerMetricVectors(RAMUsage)")
-		}
-		return nil, fmt.Errorf("GetNormalizedContainerMetricVectors(RAMUsage): %s", err)
+		return nil, prom.WrapError(err, "GetNormalizedContainerMetricVectors(RAMUsage)")
 	}
 	for key := range RAMUsedMap {
 		containers[key] = true
@@ -1780,10 +1771,7 @@ func (cm *CostModel) costDataRange(cli prometheusClient.Client, clientset kubern
 
 	CPUReqMap, err := GetNormalizedContainerMetricVectors(resCPURequests, normalizationValue, clusterID)
 	if err != nil {
-		if pce, ok := err.(prom.CommError); ok {
-			return nil, pce.Wrap("GetNormalizedContainerMetricVectors(CPURequests)")
-		}
-		return nil, fmt.Errorf("GetNormalizedContainerMetricVectors(CPURequests): %s", err)
+		return nil, prom.WrapError(err, "GetNormalizedContainerMetricVectors(CPURequests)")
 	}
 	for key := range CPUReqMap {
 		containers[key] = true
@@ -1793,10 +1781,7 @@ func (cm *CostModel) costDataRange(cli prometheusClient.Client, clientset kubern
 	// rate(container_cpu_usage_seconds_total) which properly accounts for normalized rates
 	CPUUsedMap, err := GetContainerMetricVectors(resCPUUsage, clusterID)
 	if err != nil {
-		if pce, ok := err.(prom.CommError); ok {
-			return nil, pce.Wrap("GetContainerMetricVectors(CPUUsage)")
-		}
-		return nil, fmt.Errorf("GetContainerMetricVectors(CPUUsage): %s", err)
+		return nil, prom.WrapError(err, "GetContainerMetricVectors(CPUUsage)")
 	}
 	for key := range CPUUsedMap {
 		containers[key] = true
@@ -1804,10 +1789,7 @@ func (cm *CostModel) costDataRange(cli prometheusClient.Client, clientset kubern
 
 	RAMAllocMap, err := GetContainerMetricVectors(resRAMAlloc, clusterID)
 	if err != nil {
-		if pce, ok := err.(prom.CommError); ok {
-			return nil, pce.Wrap("GetContainerMetricVectors(RAMAllocations)")
-		}
-		return nil, fmt.Errorf("GetContainerMetricVectors(RAMAllocations): %s", err)
+		return nil, prom.WrapError(err, "GetContainerMetricVectors(RAMAllocations)")
 	}
 	for key := range RAMAllocMap {
 		containers[key] = true
@@ -1815,10 +1797,7 @@ func (cm *CostModel) costDataRange(cli prometheusClient.Client, clientset kubern
 
 	CPUAllocMap, err := GetContainerMetricVectors(resCPUAlloc, clusterID)
 	if err != nil {
-		if pce, ok := err.(prom.CommError); ok {
-			return nil, pce.Wrap("GetContainerMetricVectors(CPUAllocations)")
-		}
-		return nil, fmt.Errorf("GetContainerMetricVectors(CPUAllocations): %s", err)
+		return nil, prom.WrapError(err, "GetContainerMetricVectors(CPUAllocations)")
 	}
 	for key := range CPUAllocMap {
 		containers[key] = true
@@ -1826,10 +1805,7 @@ func (cm *CostModel) costDataRange(cli prometheusClient.Client, clientset kubern
 
 	GPUReqMap, err := GetNormalizedContainerMetricVectors(resGPURequests, normalizationValue, clusterID)
 	if err != nil {
-		if pce, ok := err.(prom.CommError); ok {
-			return nil, pce.Wrap("GetContainerMetricVectors(GPURequests)")
-		}
-		return nil, fmt.Errorf("GetContainerMetricVectors(GPURequests): %s", err)
+		return nil, prom.WrapError(err, "GetContainerMetricVectors(GPURequests)")
 	}
 	for key := range GPUReqMap {
 		containers[key] = true

pkg/prom/error.go

@@ -5,19 +5,70 @@ import (
 	"strings"
 )
 
+// WrapError wraps the given error with the given message, usually for adding
+// context, but persists the existing type of error.
+func WrapError(err error, msg string) error {
+	switch e := err.(type) {
+	case CommError:
+		return e.Wrap(msg)
+	case NoDataError:
+		return e.Wrap(msg)
+	default:
+		return fmt.Errorf("%s: %s", msg, err)
+	}
+}
+
+// CommError describes an error communicating with Prometheus
 type CommError struct {
 	messages []string
 }
 
+// NewCommError creates a new CommError
 func NewCommError(messages ...string) CommError {
 	return CommError{messages: messages}
 }
 
+// IsCommError returns true if the given error is a CommError
+func IsCommError(err error) bool {
+	_, ok := err.(CommError)
+	return ok
+}
+
+// Error prints the error as a string
 func (pce CommError) Error() string {
 	return fmt.Sprintf("Prometheus communication error: %s", strings.Join(pce.messages, ": "))
 }
 
+// Wrap wraps the error with the given message, but persists the error type.
 func (pce CommError) Wrap(message string) CommError {
 	pce.messages = append([]string{message}, pce.messages...)
 	return pce
 }
+
+// NoDataError indicates that no data was returned by Prometheus. This should
+// be treated like an EOF error, in that it may be expected.
+type NoDataError struct {
+	messages []string
+}
+
+// NewNoDataError creates a new NoDataError
+func NewNoDataError(messages ...string) NoDataError {
+	return NoDataError{messages: messages}
+}
+
+// IsNoDataError returns true if the given error is a NoDataError
+func IsNoDataError(err error) bool {
+	_, ok := err.(NoDataError)
+	return ok
+}
+
+// Error prints the error as a string
+func (nde NoDataError) Error() string {
+	return fmt.Sprintf("No data error: %s", strings.Join(nde.messages, ": "))
+}
+
+// Wrap wraps the error with the given message, but persists the error type.
+func (nde NoDataError) Wrap(message string) NoDataError {
+	nde.messages = append([]string{message}, nde.messages...)
+	return nde
+}

pkg/prom/result.go

@@ -33,7 +33,7 @@ func MetricFieldFormatErr(query string) error {
 }
 
 func NoDataErr(query string) error {
-	return fmt.Errorf("No data fetching query '%s'", query)
+	return NewNoDataError(query)
 }
 
 func PromUnexpectedResponseErr(query string) error {
@@ -41,7 +41,7 @@ func PromUnexpectedResponseErr(query string) error {
 }
 
 func QueryResultNilErr(query string) error {
-	return NewCommError(fmt.Sprintf("nil queryResult fetching query '%s'", query))
+	return NewCommError(query)
 }
 
 func ResultFieldDoesNotExistErr(query string) error {