For the series() call don't fetch all the data too

Workaround for upstream issue prometheus/prometheus#4057

Author: jacksontj

promclient/client.go

@@ -48,13 +48,26 @@ func GetData(ctx context.Context, url string, client *http.Client, labelset mode
 	}
 }
 
-func GetSeries(ctx context.Context, url string, client *http.Client) ([]model.LabelSet, error) {
+func GetSeries(ctx context.Context, url string, client *http.Client, labelset model.LabelSet) (model.Value, error) {
 	promResp := &SeriesResult{}
 	if err := DoRequest(ctx, url, client, promResp); err == nil {
 		if promResp.Status != promhttputil.StatusSuccess {
 			return nil, fmt.Errorf(promResp.Error)
 		}
-		return promResp.Data, nil
+
+		// convert to vector (there aren't points, but this way we don't have to make more merging functions)
+		retVector := make(model.Vector, len(promResp.Data))
+		for j, labelset := range promResp.Data {
+			retVector[j] = &model.Sample{
+				Metric: model.Metric(labelset),
+			}
+		}
+
+		if err := promhttputil.ValueAddLabelSet(retVector, labelset); err != nil {
+			return nil, err
+		}
+
+		return retVector, nil
 	} else {
 		return nil, err
 	}

proxyquerier/querier.go

@@ -8,6 +8,7 @@ import (
 	"github.com/jacksontj/promxy/promclient"
 	"github.com/jacksontj/promxy/servergroup"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/pkg/labels"
 	"github.com/prometheus/prometheus/pkg/timestamp"
 	"github.com/prometheus/prometheus/storage"
@@ -45,7 +46,18 @@ func (h *ProxyQuerier) Select(selectParams *storage.SelectParams, matchers ...*l
 		}).Debug("Select")
 	}()
 
-	result, err := h.ServerGroups.GetValue(h.Ctx, timestamp.Time(selectParams.Start), timestamp.Time(selectParams.End), matchers)
+	var result model.Value
+	var err error
+	// Select() is a combined API call for query/query_range/series.
+	// as of right now there is no great way of differentiating between a
+	// data call (query/query_range) and a metadata call (series). For now
+	// the working workaround is to switch based on the selectParams.
+	// https://github.com/prometheus/prometheus/issues/4057
+	if selectParams == nil {
+		result, err = h.ServerGroups.GetSeries(h.Ctx, h.Start, h.End, matchers)
+	} else {
+		result, err = h.ServerGroups.GetValue(h.Ctx, timestamp.Time(selectParams.Start), timestamp.Time(selectParams.End), matchers)
+	}
 	if err != nil {
 		return nil, err
 	}

servergroup/servergroup.go

@@ -460,3 +460,90 @@ func (s *ServerGroup) GetValuesForLabelName(ctx context.Context, path string) ([
 
 	return result, nil
 }
+
+func (s *ServerGroup) GetSeries(ctx context.Context, start, end time.Time, matchers []*labels.Matcher) (model.Value, error) {
+	filteredMatchers, ok := s.FilterMatchers(matchers)
+	if !ok {
+		return nil, nil
+	}
+
+	// Create the query params
+	values := url.Values{}
+
+	urlBase := "/api/v1/series"
+
+	// Add matchers
+	// http://localhost:8080/api/v1/query?query=scrape_duration_seconds%7Bjob%3D%22prometheus%22%7D&time=1507412244.663&_=1507412096887
+	pql, err := promhttputil.MatcherToString(filteredMatchers)
+	if err != nil {
+		return nil, err
+	}
+	values.Add("match[]", pql)
+
+	values.Add("start", model.Time(timestamp.FromTime(start)).String())
+	values.Add("end", model.Time(timestamp.FromTime(end)).String())
+
+	targets := s.Targets()
+
+	childContext, childContextCancel := context.WithCancel(ctx)
+	defer childContextCancel()
+	resultChans := make([]chan interface{}, len(targets))
+
+	for i, target := range targets {
+		resultChans[i] = make(chan interface{}, 1)
+		parsedUrl, err := url.Parse(target + urlBase)
+		if err != nil {
+			return nil, err
+		}
+		parsedUrl.RawQuery = values.Encode()
+		go func(retChan chan interface{}, stringUrl string) {
+			start := time.Now()
+			result, err := promclient.GetSeries(childContext, stringUrl, s.Client, s.Cfg.Labels)
+			took := time.Now().Sub(start)
+			if err != nil {
+				serverGroupSummary.WithLabelValues(parsedUrl.Host, "getdata", "error").Observe(float64(took))
+				retChan <- err
+			} else {
+				serverGroupSummary.WithLabelValues(parsedUrl.Host, "getdata", "success").Observe(float64(took))
+				retChan <- result
+			}
+		}(resultChans[i], parsedUrl.String())
+	}
+
+	// Wait for results as we get them
+	var result model.Value
+	var lastError error
+	errCount := 0
+	for i := 0; i < len(targets); i++ {
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+
+		case ret := <-resultChans[i]:
+			switch retTyped := ret.(type) {
+			case error:
+				lastError = retTyped
+				errCount++
+			case model.Value:
+				// TODO: check qData.ResultType
+				if result == nil {
+					result = retTyped
+				} else {
+					var err error
+					result, err = promhttputil.MergeValues(s.Cfg.GetAntiAffinity(), result, retTyped)
+					if err != nil {
+						return nil, err
+					}
+				}
+			default:
+				return nil, fmt.Errorf("Unknown return type")
+			}
+		}
+	}
+
+	if errCount != 0 && errCount == len(targets) {
+		return nil, fmt.Errorf("Unable to fetch from downstream servers, lastError: %s", lastError.Error())
+	}
+
+	return result, nil
+}

servergroup/servergroups.go

@@ -163,3 +163,53 @@ func (s ServerGroups) GetValuesForLabelName(ctx context.Context, path string) ([
 
 	return result, nil
 }
+
+func (s ServerGroups) GetSeries(ctx context.Context, start, end time.Time, matchers []*labels.Matcher) (model.Value, error) {
+	childContext, childContextCancel := context.WithCancel(ctx)
+	defer childContextCancel()
+	resultChans := make([]chan interface{}, len(s))
+
+	// Scatter out all the queries
+	for i, serverGroup := range s {
+		resultChans[i] = make(chan interface{}, 1)
+		go func(retChan chan interface{}, serverGroup *ServerGroup) {
+			result, err := serverGroup.GetSeries(childContext, start, end, matchers)
+			if err != nil {
+				retChan <- err
+			} else {
+				retChan <- result
+			}
+		}(resultChans[i], serverGroup)
+	}
+
+	// Wait for results as we get them
+	var result model.Value
+	var lastError error
+	errCount := 0
+	for i := 0; i < len(s); i++ {
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		case ret := <-resultChans[i]:
+			switch retTyped := ret.(type) {
+			case error:
+				lastError = retTyped
+				errCount++
+			case model.Value:
+				var err error
+				result, err = promhttputil.MergeValues(model.TimeFromUnix(0), result, retTyped)
+				if err != nil {
+					return nil, err
+				}
+			}
+		}
+
+	}
+
+	// If we got only errors, lets return that
+	if errCount == len(s) {
+		return nil, fmt.Errorf("Unable to fetch from downstream servers, lastError: %s", lastError.Error())
+	}
+
+	return result, nil
+}