extracting prometheus query execution to util

truerocha · Apr 16, 2019 · d7f81e7 · d7f81e7
1 parent 016b0da
commit d7f81e7
Show file tree

Hide file tree

Showing 2 changed files with 52 additions and 46 deletions.
diff --git a/clusterloader2/pkg/measurement/common/slos/api_responsiveness_prometheus.go b/clusterloader2/pkg/measurement/common/slos/api_responsiveness_prometheus.go
@@ -29,7 +29,6 @@ import (
 
 	"github.com/prometheus/common/model"
 	"k8s.io/apimachinery/pkg/util/sets"
-	"k8s.io/apimachinery/pkg/util/wait"
 	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/klog"
 	"k8s.io/perf-tests/clusterloader2/pkg/errors"
@@ -47,8 +46,6 @@ const (
 	countQuery = "sum(increase(apiserver_request_duration_seconds_count[%v])) by (resource, subresource, scope, verb)"
 
 	latencyWindowSize = 5 * time.Minute
-	queryTimeout      = 5 * time.Minute
-	queryInterval     = 30 * time.Second
 )
 
 func init() {
@@ -151,12 +148,12 @@ func (a *apiResponsivenessMeasurementPrometheus) gatherApiCalls(c clientset.Inte
 		latencyMeasurementDuration = time.Minute
 	}
 	timeBoundedLatencyQuery := fmt.Sprintf(latencyQuery, measurementutil.ToPrometheusTime(latencyMeasurementDuration))
-	latencySamples, err := gatherSamples(c, timeBoundedLatencyQuery, measurementEnd)
+	latencySamples, err := measurementutil.ExecutePrometheusQuery(c, timeBoundedLatencyQuery, measurementEnd)
 	if err != nil {
 		return nil, err
 	}
 	timeBoundedCountQuery := fmt.Sprintf(countQuery, measurementutil.ToPrometheusTime(measurementDuration))
-	countSamples, err := gatherSamples(c, timeBoundedCountQuery, measurementEnd)
+	countSamples, err := measurementutil.ExecutePrometheusQuery(c, timeBoundedCountQuery, measurementEnd)
 	if err != nil {
 		return nil, err
 	}
@@ -234,47 +231,6 @@ func addCount(apiCalls map[string]*apiCall, resource, subresource, verb, scope s
 	call.Count = count
 }
 
-func gatherSamples(c clientset.Interface, query string, queryTime time.Time) ([]*model.Sample, error) {
-	if queryTime.IsZero() {
-		return nil, fmt.Errorf("query time can't be zero")
-	}
-
-	var body []byte
-	var queryErr error
-	params := map[string]string{
-		"query": query,
-		"time":  queryTime.Format(time.RFC3339),
-	}
-	if err := wait.PollImmediate(queryInterval, queryTimeout, func() (bool, error) {
-		body, queryErr = c.CoreV1().
-			Services("monitoring").
-			ProxyGet("http", "prometheus-k8s", "9090", "api/v1/query", params).
-			DoRaw()
-		if queryErr != nil {
-			return false, nil
-		}
-		return true, nil
-	}); err != nil {
-		if queryErr != nil {
-			return nil, fmt.Errorf("query error: %v", queryErr)
-		}
-		return nil, fmt.Errorf("query error: %v", err)
-	}
-
-	samples, err := measurementutil.ExtractMetricSamples2(body)
-	if err != nil {
-		return nil, fmt.Errorf("exctracting error: %v", err)
-	}
-
-	var resultSamples []*model.Sample
-	for _, sample := range samples {
-		if !math.IsNaN(float64(sample.Value)) {
-			resultSamples = append(resultSamples, sample)
-		}
-	}
-	return resultSamples, nil
-}
-
 func getMetricKey(resource, subresource, verb, scope string) string {
 	return fmt.Sprintf("%s|%s|%s|%s", resource, subresource, verb, scope)
 }

diff --git a/clusterloader2/pkg/measurement/util/prometheus.go b/clusterloader2/pkg/measurement/util/prometheus.go
@@ -20,11 +20,19 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"math"
 	"strings"
 	"time"
 
 	"github.com/prometheus/common/expfmt"
 	"github.com/prometheus/common/model"
+	"k8s.io/apimachinery/pkg/util/wait"
+	clientset "k8s.io/client-go/kubernetes"
+)
+
+const (
+	queryTimeout  = 5 * time.Minute
+	queryInterval = 30 * time.Second
 )
 
 // ExtractMetricSamples unpacks metric blob into prometheus model structures.
@@ -74,6 +82,48 @@ type promResponseData struct {
 	v model.Value
 }
 
+// ExecutePrometheusQuery executes given prometheus query.
+func ExecutePrometheusQuery(c clientset.Interface, query string, queryTime time.Time) ([]*model.Sample, error) {
+	if queryTime.IsZero() {
+		return nil, fmt.Errorf("query time can't be zero")
+	}
+
+	var body []byte
+	var queryErr error
+	params := map[string]string{
+		"query": query,
+		"time":  queryTime.Format(time.RFC3339),
+	}
+	if err := wait.PollImmediate(queryInterval, queryTimeout, func() (bool, error) {
+		body, queryErr = c.CoreV1().
+			Services("monitoring").
+			ProxyGet("http", "prometheus-k8s", "9090", "api/v1/query", params).
+			DoRaw()
+		if queryErr != nil {
+			return false, nil
+		}
+		return true, nil
+	}); err != nil {
+		if queryErr != nil {
+			return nil, fmt.Errorf("query error: %v", queryErr)
+		}
+		return nil, fmt.Errorf("query error: %v", err)
+	}
+
+	samples, err := ExtractMetricSamples2(body)
+	if err != nil {
+		return nil, fmt.Errorf("exctracting error: %v", err)
+	}
+
+	var resultSamples []*model.Sample
+	for _, sample := range samples {
+		if !math.IsNaN(float64(sample.Value)) {
+			resultSamples = append(resultSamples, sample)
+		}
+	}
+	return resultSamples, nil
+}
+
 // UnmarshalJSON unmarshals json into promResponseData structure.
 func (qr *promResponseData) UnmarshalJSON(b []byte) error {
 	v := struct {