Merge pull request kubernetes#3122 from yashbhutwala/history-resolution

Add history resolution and use prometheus client

Author: k8s-ci-robot

vertical-pod-autoscaler/go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/golang/mock v1.3.1
 	github.com/prometheus/client_golang v0.9.2
 	github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4 // indirect
-	github.com/prometheus/common v0.4.1 // indirect
+	github.com/prometheus/common v0.4.1
 	github.com/prometheus/procfs v0.0.5 // indirect
 	github.com/stretchr/testify v1.4.0
 	golang.org/x/crypto v0.0.0-20200221231518-2aa609cf4a9d // indirect

vertical-pod-autoscaler/pkg/recommender/input/history/history_provider.go

@@ -17,23 +17,30 @@ limitations under the License.
 package history
 
 import (
+	"context"
 	"fmt"
-	"net/http"
+	"k8s.io/klog"
 	"sort"
 	"strings"
 	"time"
 
+	promapi "github.com/prometheus/client_golang/api"
+	prometheusv1 "github.com/prometheus/client_golang/api/prometheus/v1"
+	prommodel "github.com/prometheus/common/model"
+
 	"k8s.io/autoscaler/vertical-pod-autoscaler/pkg/recommender/model"
 )
 
 // PrometheusHistoryProviderConfig allow to select which metrics
 // should be queried to get real resource utilization.
 type PrometheusHistoryProviderConfig struct {
-	Address                                            string
-	HistoryLength, PodLabelPrefix, PodLabelsMetricName string
-	PodNamespaceLabel, PodNameLabel                    string
-	CtrNamespaceLabel, CtrPodNameLabel, CtrNameLabel   string
-	CadvisorMetricsJobName                             string
+	Address                                          string
+	QueryTimeout                                     time.Duration
+	HistoryLength, HistoryResolution                 string
+	PodLabelPrefix, PodLabelsMetricName              string
+	PodNamespaceLabel, PodNameLabel                  string
+	CtrNamespaceLabel, CtrPodNameLabel, CtrNameLabel string
+	CadvisorMetricsJobName                           string
 }
 
 // PodHistory represents history of usage and labels for a given pod.
@@ -58,19 +65,44 @@ type HistoryProvider interface {
 }
 
 type prometheusHistoryProvider struct {
-	prometheusClient PrometheusClient
-	config           PrometheusHistoryProviderConfig
+	prometheusClient  prometheusv1.API
+	config            PrometheusHistoryProviderConfig
+	queryTimeout      time.Duration
+	historyDuration   prommodel.Duration
+	historyResolution prommodel.Duration
 }
 
 // NewPrometheusHistoryProvider contructs a history provider that gets data from Prometheus.
-func NewPrometheusHistoryProvider(config PrometheusHistoryProviderConfig) HistoryProvider {
-	return &prometheusHistoryProvider{
-		prometheusClient: NewPrometheusClient(&http.Client{}, config.Address),
-		config:           config,
+func NewPrometheusHistoryProvider(config PrometheusHistoryProviderConfig) (HistoryProvider, error) {
+	promClient, err := promapi.NewClient(promapi.Config{
+		Address: config.Address,
+	})
+	if err != nil {
+		return &prometheusHistoryProvider{}, err
+	}
+
+	// Use Prometheus's model.Duration; this can additionally parse durations in days, weeks and years (as well as seconds, minutes, hours etc)
+	historyDuration, err := prommodel.ParseDuration(config.HistoryLength)
+	if err != nil {
+		return &prometheusHistoryProvider{}, fmt.Errorf("history length %s is not a valid Prometheus duration: %v", config.HistoryLength, err)
+	}
+
+	historyResolution, err := prommodel.ParseDuration(config.HistoryResolution)
+	if err != nil {
+		return &prometheusHistoryProvider{}, fmt.Errorf("history resolution %s is not a valid Prometheus duration: %v", config.HistoryResolution, err)
 	}
+
+	return &prometheusHistoryProvider{
+		prometheusClient:  prometheusv1.NewAPI(promClient),
+		config:            config,
+		queryTimeout:      config.QueryTimeout,
+		historyDuration:   historyDuration,
+		historyResolution: historyResolution,
+	}, nil
 }
 
-func (p *prometheusHistoryProvider) getContainerIDFromLabels(labels map[string]string) (*model.ContainerID, error) {
+func (p *prometheusHistoryProvider) getContainerIDFromLabels(metric prommodel.Metric) (*model.ContainerID, error) {
+	labels := promMetricToLabelMap(metric)
 	namespace, ok := labels[p.config.CtrNamespaceLabel]
 	if !ok {
 		return nil, fmt.Errorf("no %s label", p.config.CtrNamespaceLabel)
@@ -86,13 +118,12 @@ func (p *prometheusHistoryProvider) getContainerIDFromLabels(labels map[string]s
 	return &model.ContainerID{
 		PodID: model.PodID{
 			Namespace: namespace,
-			PodName:   podName,
-		},
-		ContainerName: containerName,
-	}, nil
+			PodName:   podName},
+		ContainerName: containerName}, nil
 }
 
-func (p *prometheusHistoryProvider) getPodIDFromLabels(labels map[string]string) (*model.PodID, error) {
+func (p *prometheusHistoryProvider) getPodIDFromLabels(metric prommodel.Metric) (*model.PodID, error) {
+	labels := promMetricToLabelMap(metric)
 	namespace, ok := labels[p.config.PodNamespaceLabel]
 	if !ok {
 		return nil, fmt.Errorf("no %s label", p.config.PodNamespaceLabel)
@@ -104,17 +135,25 @@ func (p *prometheusHistoryProvider) getPodIDFromLabels(labels map[string]string)
 	return &model.PodID{Namespace: namespace, PodName: podName}, nil
 }
 
-func (p *prometheusHistoryProvider) getPodLabelsMap(metricLabels map[string]string) map[string]string {
+func (p *prometheusHistoryProvider) getPodLabelsMap(metric prommodel.Metric) map[string]string {
 	podLabels := make(map[string]string)
-	for key, value := range metricLabels {
-		podLabelKey := strings.TrimPrefix(key, p.config.PodLabelPrefix)
-		if podLabelKey != key {
-			podLabels[podLabelKey] = value
+	for key, value := range metric {
+		podLabelKey := strings.TrimPrefix(string(key), p.config.PodLabelPrefix)
+		if podLabelKey != string(key) {
+			podLabels[podLabelKey] = string(value)
 		}
 	}
 	return podLabels
 }
 
+func promMetricToLabelMap(metric prommodel.Metric) map[string]string {
+	labels := map[string]string{}
+	for k, v := range metric {
+		labels[string(k)] = string(v)
+	}
+	return labels
+}
+
 func resourceAmountFromValue(value float64, resource model.ResourceName) model.ResourceAmount {
 	// This assumes CPU value is in cores and memory in bytes, which is true
 	// for the metrics this class queries from Prometheus.
@@ -127,29 +166,46 @@ func resourceAmountFromValue(value float64, resource model.ResourceName) model.R
 	return model.ResourceAmount(0)
 }
 
-func getContainerUsageSamplesFromSamples(samples []Sample, resource model.ResourceName) []model.ContainerUsageSample {
+func getContainerUsageSamplesFromSamples(samples []prommodel.SamplePair, resource model.ResourceName) []model.ContainerUsageSample {
 	res := make([]model.ContainerUsageSample, 0)
 	for _, sample := range samples {
 		res = append(res, model.ContainerUsageSample{
-			MeasureStart: sample.Timestamp,
-			Usage:        resourceAmountFromValue(sample.Value, resource),
-			Resource:     resource,
-		})
+			MeasureStart: sample.Timestamp.Time(),
+			Usage:        resourceAmountFromValue(float64(sample.Value), resource),
+			Resource:     resource})
 	}
 	return res
 }
 
 func (p *prometheusHistoryProvider) readResourceHistory(res map[model.PodID]*PodHistory, query string, resource model.ResourceName) error {
-	tss, err := p.prometheusClient.GetTimeseries(query)
+	end := time.Now()
+	start := end.Add(-time.Duration(p.historyDuration))
+
+	ctx, cancel := context.WithTimeout(context.Background(), p.queryTimeout)
+	defer cancel()
+
+	result, err := p.prometheusClient.QueryRange(ctx, query, prometheusv1.Range{
+		Start: start,
+		End:   end,
+		Step:  time.Duration(p.historyResolution),
+	})
+
 	if err != nil {
 		return fmt.Errorf("cannot get timeseries for %v: %v", resource, err)
 	}
-	for _, ts := range tss {
-		containerID, err := p.getContainerIDFromLabels(ts.Labels)
+
+	matrix, ok := result.(prommodel.Matrix)
+	if !ok {
+		return fmt.Errorf("expected query to return a matrix; got result type %T", result)
+	}
+
+	for _, ts := range matrix {
+		containerID, err := p.getContainerIDFromLabels(ts.Metric)
 		if err != nil {
-			return fmt.Errorf("cannot get container ID from labels: %v", ts.Labels)
+			return fmt.Errorf("cannot get container ID from labels: %v", ts.Metric)
 		}
-		newSamples := getContainerUsageSamplesFromSamples(ts.Samples, resource)
+
+		newSamples := getContainerUsageSamplesFromSamples(ts.Values, resource)
 		podHistory, ok := res[containerID.PodID]
 		if !ok {
 			podHistory = newEmptyHistory()
@@ -163,26 +219,37 @@ func (p *prometheusHistoryProvider) readResourceHistory(res map[model.PodID]*Pod
 }
 
 func (p *prometheusHistoryProvider) readLastLabels(res map[model.PodID]*PodHistory, query string) error {
-	tss, err := p.prometheusClient.GetTimeseries(query)
+	ctx, cancel := context.WithTimeout(context.Background(), p.queryTimeout)
+	defer cancel()
+
+	result, err := p.prometheusClient.Query(ctx, query, time.Now())
 	if err != nil {
 		return fmt.Errorf("cannot get timeseries for labels: %v", err)
 	}
-	for _, ts := range tss {
-		podID, err := p.getPodIDFromLabels(ts.Labels)
+
+	matrix, ok := result.(prommodel.Matrix)
+	if !ok {
+		return fmt.Errorf("expected query to return a matrix; got result type %T", result)
+	}
+
+	for _, ts := range matrix {
+		podID, err := p.getPodIDFromLabels(ts.Metric)
 		if err != nil {
-			return fmt.Errorf("cannot get container ID from labels: %v", ts.Labels)
+			return fmt.Errorf("cannot get container ID from labels %v: %v", ts.Metric, err)
 		}
 		podHistory, ok := res[*podID]
 		if !ok {
 			podHistory = newEmptyHistory()
 			res[*podID] = podHistory
 		}
-		podLabels := p.getPodLabelsMap(ts.Labels)
-		for _, sample := range ts.Samples {
-			if sample.Timestamp.After(podHistory.LastSeen) {
-				podHistory.LastSeen = sample.Timestamp
-				podHistory.LastLabels = podLabels
-			}
+		podLabels := p.getPodLabelsMap(ts.Metric)
+
+		// time series results will always be sorted chronologically from oldest to
+		// newest, so the last element is the latest sample
+		lastSample := ts.Values[len(ts.Values)-1]
+		if lastSample.Timestamp.Time().After(podHistory.LastSeen) {
+			podHistory.LastSeen = lastSample.Timestamp.Time()
+			podHistory.LastLabels = podLabels
 		}
 	}
 	return nil
@@ -197,12 +264,16 @@ func (p *prometheusHistoryProvider) GetClusterHistory() (map[model.PodID]*PodHis
 	podSelector = podSelector + fmt.Sprintf("%s=~\".+\", %s!=\"POD\", %s!=\"\"",
 		p.config.CtrPodNameLabel, p.config.CtrNameLabel, p.config.CtrNameLabel)
 
-	// This query uses Prometheus Subquery notation, to gives us a result of a five minute cpu rate by default evaluated every 1minute for last config.HistoryLength days/hours/minutes. In order to change the evaluation step, you need change Prometheus global.evaluation_interval configuration parameter.
-	err := p.readResourceHistory(res, fmt.Sprintf("rate(container_cpu_usage_seconds_total{%s}[5m])[%s:]", podSelector, p.config.HistoryLength), model.ResourceCPU)
+	historicalCpuQuery := fmt.Sprintf("rate(container_cpu_usage_seconds_total{%s}[%s])", podSelector, p.config.HistoryResolution)
+	klog.V(4).Infof("Historical CPU usage query used: %s", historicalCpuQuery)
+	err := p.readResourceHistory(res, historicalCpuQuery, model.ResourceCPU)
 	if err != nil {
 		return nil, fmt.Errorf("cannot get usage history: %v", err)
 	}
-	err = p.readResourceHistory(res, fmt.Sprintf("container_memory_working_set_bytes{%s}[%s]", podSelector, p.config.HistoryLength), model.ResourceMemory)
+
+	historicalMemoryQuery := fmt.Sprintf("container_memory_working_set_bytes{%s}", podSelector)
+	klog.V(4).Infof("Historical memory usage query used: %s", historicalMemoryQuery)
+	err = p.readResourceHistory(res, historicalMemoryQuery, model.ResourceMemory)
 	if err != nil {
 		return nil, fmt.Errorf("cannot get usage history: %v", err)
 	}
@@ -211,6 +282,6 @@ func (p *prometheusHistoryProvider) GetClusterHistory() (map[model.PodID]*PodHis
 			sort.Slice(samples, func(i, j int) bool { return samples[i].MeasureStart.Before(samples[j].MeasureStart) })
 		}
 	}
-	p.readLastLabels(res, fmt.Sprintf("%s[%s]", p.config.PodLabelsMetricName, p.config.HistoryLength))
+	p.readLastLabels(res, p.config.PodLabelsMetricName)
 	return res, nil
 }