Other adjustments to new packages and types

Author: juliusv

pkg/chunk/aws_storage_client.go

@@ -2,6 +2,7 @@ package chunk
 
 import (
 	"bytes"
+	"context"
 	"flag"
 	"fmt"
 	"io/ioutil"
@@ -21,7 +22,6 @@ import (
 	"github.com/aws/aws-sdk-go/service/s3/s3iface"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/log"
-	"golang.org/x/net/context"
 
 	awscommon "github.com/weaveworks/common/aws"
 	"github.com/weaveworks/common/instrument"

pkg/chunk/aws_storage_client_test.go

@@ -2,6 +2,7 @@ package chunk
 
 import (
 	"bytes"
+	"context"
 	"fmt"
 	"io/ioutil"
 	"math/rand"
@@ -22,7 +23,6 @@ import (
 	"github.com/prometheus/common/log"
 	"github.com/prometheus/common/model"
 	"github.com/stretchr/testify/require"
-	"golang.org/x/net/context"
 
 	"github.com/weaveworks/cortex/pkg/util"
 )

pkg/chunk/chunk.go

@@ -13,8 +13,7 @@ import (
 	"github.com/golang/snappy"
 	"github.com/pkg/errors"
 	"github.com/prometheus/common/model"
-	"github.com/prometheus/prometheus/storage/local"
-	prom_chunk "github.com/prometheus/prometheus/storage/local/chunk"
+	prom_chunk "github.com/weaveworks/cortex/pkg/prom1/storage/local/chunk"
 
 	errs "github.com/weaveworks/common/errors"
 	"github.com/weaveworks/cortex/pkg/util"
@@ -280,7 +279,7 @@ func (c *Chunk) Decode(input []byte) error {
 	})
 }
 
-func chunksToIterators(chunks []Chunk) ([]local.SeriesIterator, error) {
+func chunksToMatrix(chunks []Chunk) (model.Matrix, error) {
 	// Group chunks by series, sort and dedupe samples.
 	sampleStreams := map[model.Fingerprint]*model.SampleStream{}
 	for _, c := range chunks {
@@ -301,12 +300,15 @@ func chunksToIterators(chunks []Chunk) ([]local.SeriesIterator, error) {
 		ss.Values = util.MergeSampleSets(ss.Values, samples)
 	}
 
-	iterators := make([]local.SeriesIterator, 0, len(sampleStreams))
+	matrix := make(model.Matrix, 0, len(sampleStreams))
 	for _, ss := range sampleStreams {
-		iterators = append(iterators, util.NewSampleStreamIterator(ss))
+		matrix = append(matrix, &model.SampleStream{
+			Metric: ss.Metric,
+			Values: ss.Values,
+		})
 	}
 
-	return iterators, nil
+	return matrix, nil
 }
 
 // Samples returns all SamplePairs for the chunk.

pkg/chunk/chunk_cache.go

@@ -1,6 +1,7 @@
 package chunk
 
 import (
+	"context"
 	"flag"
 	"sync"
 	"time"
@@ -9,7 +10,6 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/log"
 	"github.com/weaveworks/common/instrument"
-	"golang.org/x/net/context"
 
 	"github.com/weaveworks/cortex/pkg/util"
 )

pkg/chunk/chunk_cache_test.go

@@ -7,8 +7,8 @@ import (
 
 	"github.com/bradfitz/gomemcache/memcache"
 	"github.com/prometheus/common/model"
-	"github.com/prometheus/prometheus/storage/local/chunk"
 	"github.com/stretchr/testify/require"
+	"github.com/weaveworks/cortex/pkg/prom1/storage/local/chunk"
 	"golang.org/x/net/context"
 )
 

pkg/chunk/chunk_store.go

@@ -1,17 +1,16 @@
 package chunk
 
 import (
+	"context"
 	"encoding/json"
 	"flag"
 	"fmt"
 	"sort"
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/model"
+	"github.com/prometheus/prometheus/pkg/labels"
 	"github.com/prometheus/prometheus/promql"
-	"github.com/prometheus/prometheus/storage/local"
-	"github.com/prometheus/prometheus/storage/metric"
-	"golang.org/x/net/context"
 
 	"github.com/weaveworks/common/user"
 	"github.com/weaveworks/cortex/pkg/util"
@@ -150,7 +149,7 @@ func (c *Store) calculateDynamoWrites(userID string, chunks []Chunk) (WriteBatch
 }
 
 // Get implements ChunkStore
-func (c *Store) Get(ctx context.Context, from, through model.Time, allMatchers ...*metric.LabelMatcher) ([]local.SeriesIterator, error) {
+func (c *Store) Get(ctx context.Context, from, through model.Time, allMatchers ...*labels.Matcher) (model.Matrix, error) {
 	if through < from {
 		return nil, fmt.Errorf("invalid query, through < from (%d < %d)", through, from)
 	}
@@ -159,7 +158,7 @@ func (c *Store) Get(ctx context.Context, from, through model.Time, allMatchers .
 	if from.After(now) {
 		// time-span start is in future ... regard as legal
 		util.WithContext(ctx).Debugf("Whole timerange %v..%v in future (now=%v) yield empty resultset", through, from, now)
-		return []local.SeriesIterator{}, nil
+		return nil, nil
 	}
 
 	if through.After(now) {
@@ -170,23 +169,23 @@ func (c *Store) Get(ctx context.Context, from, through model.Time, allMatchers .
 
 	// Fetch metric name chunks if the matcher is of type equal,
 	metricNameMatcher, matchers, ok := util.ExtractMetricNameMatcherFromMatchers(allMatchers)
-	if ok && metricNameMatcher.Type == metric.Equal {
-		return c.getMetricNameIterators(ctx, from, through, matchers, metricNameMatcher.Value)
+	if ok && metricNameMatcher.Type == labels.MatchEqual {
+		return c.getMetricNameMatrix(ctx, from, through, matchers, metricNameMatcher.Value)
 	}
 
 	// Otherwise we will create lazy iterators for all series in our index
-	return c.getSeriesIterators(ctx, from, through, matchers, metricNameMatcher)
+	return c.getSeriesMatrix(ctx, from, through, matchers, metricNameMatcher)
 }
 
-func (c *Store) getMetricNameIterators(ctx context.Context, from, through model.Time, allMatchers []*metric.LabelMatcher, metricName model.LabelValue) ([]local.SeriesIterator, error) {
+func (c *Store) getMetricNameMatrix(ctx context.Context, from, through model.Time, allMatchers []*labels.Matcher, metricName string) (model.Matrix, error) {
 	chunks, err := c.getMetricNameChunks(ctx, from, through, allMatchers, metricName)
 	if err != nil {
 		return nil, err
 	}
-	return chunksToIterators(chunks)
+	return chunksToMatrix(chunks)
 }
 
-func (c *Store) getMetricNameChunks(ctx context.Context, from, through model.Time, allMatchers []*metric.LabelMatcher, metricName model.LabelValue) ([]Chunk, error) {
+func (c *Store) getMetricNameChunks(ctx context.Context, from, through model.Time, allMatchers []*labels.Matcher, metricName string) ([]Chunk, error) {
 	logger := util.WithContext(ctx)
 	filters, matchers := util.SplitFiltersAndMatchers(allMatchers)
 	chunks, err := c.lookupChunksByMetricName(ctx, from, through, matchers, metricName)
@@ -228,7 +227,7 @@ func (c *Store) getMetricNameChunks(ctx context.Context, from, through model.Tim
 outer:
 	for _, chunk := range allChunks {
 		for _, filter := range filters {
-			if !filter.Match(chunk.Metric[filter.Name]) {
+			if !filter.Matches(string(chunk.Metric[model.LabelName(filter.Name)])) {
 				continue outer
 			}
 		}
@@ -238,7 +237,13 @@ outer:
 	return filteredChunks, nil
 }
 
-func (c *Store) getSeriesIterators(ctx context.Context, from, through model.Time, allMatchers []*metric.LabelMatcher, metricNameMatcher *metric.LabelMatcher) ([]local.SeriesIterator, error) {
+type byMatcherLabel []*labels.Matcher
+
+func (lms byMatcherLabel) Len() int           { return len(lms) }
+func (lms byMatcherLabel) Swap(i, j int)      { lms[i], lms[j] = lms[j], lms[i] }
+func (lms byMatcherLabel) Less(i, j int) bool { return lms[i].Name < lms[j].Name }
+
+func (c *Store) getSeriesMatrix(ctx context.Context, from, through model.Time, allMatchers []*labels.Matcher, metricNameMatcher *labels.Matcher) (model.Matrix, error) {
 	// Get all series from the index
 	userID, err := user.ExtractOrgID(ctx)
 	if err != nil {
@@ -253,7 +258,7 @@ func (c *Store) getSeriesIterators(ctx context.Context, from, through model.Time
 		return nil, err
 	}
 
-	lazyIterators := make([]local.SeriesIterator, 0, len(seriesEntries))
+	matrix := make(model.Matrix, 0, len(seriesEntries))
 outer:
 	for _, seriesEntry := range seriesEntries {
 		metric, err := parseSeriesRangeValue(seriesEntry.RangeValue, seriesEntry.Value)
@@ -262,40 +267,52 @@ outer:
 		}
 
 		// Apply metric name matcher
-		if metricNameMatcher != nil && !metricNameMatcher.Match(metric[metricNameMatcher.Name]) {
+		if metricNameMatcher != nil && !metricNameMatcher.Matches(string(metric[model.LabelName(metricNameMatcher.Name)])) {
 			continue outer
 		}
 
 		// Apply matchers
 		for _, matcher := range allMatchers {
-			if !matcher.Match(metric[matcher.Name]) {
+			if !matcher.Matches(string(metric[model.LabelName(matcher.Name)])) {
 				continue outer
 			}
 		}
 
-		orgID, err := user.ExtractOrgID(ctx)
-		if err != nil {
-			return nil, err
+		// TODO(prom2): Does this contraption over-match?
+		var matchers []*labels.Matcher
+		for labelName, labelValue := range metric {
+			if labelName == "__name__" {
+				continue
+			}
+
+			matcher, err := labels.NewMatcher(labels.MatchEqual, string(labelName), string(labelValue))
+			if err != nil {
+				return nil, err
+			}
+			matchers = append(matchers, matcher)
 		}
-		newIterator, err := NewLazySeriesIterator(c, metric, from, through, orgID)
+		// TODO(prom2): why is sorting needed?
+		sort.Sort(byMatcherLabel(matchers))
+
+		m, err := c.getMetricNameMatrix(ctx, from, through, matchers, metricNameMatcher.Name)
 		if err != nil {
 			return nil, err
 		}
 
-		lazyIterators = append(lazyIterators, newIterator)
+		matrix = append(matrix, m...)
 	}
-	return lazyIterators, nil
+	return matrix, nil
 }
 
-func (c *Store) lookupChunksByMetricName(ctx context.Context, from, through model.Time, matchers []*metric.LabelMatcher, metricName model.LabelValue) ([]Chunk, error) {
+func (c *Store) lookupChunksByMetricName(ctx context.Context, from, through model.Time, matchers []*labels.Matcher, metricName string) ([]Chunk, error) {
 	userID, err := user.ExtractOrgID(ctx)
 	if err != nil {
 		return nil, err
 	}
 
 	// Just get chunks for metric if there are no matchers
 	if len(matchers) == 0 {
-		queries, err := c.schema.GetReadQueriesForMetric(from, through, userID, metricName)
+		queries, err := c.schema.GetReadQueriesForMetric(from, through, userID, model.LabelValue(metricName))
 		if err != nil {
 			return nil, err
 		}
@@ -312,14 +329,14 @@ func (c *Store) lookupChunksByMetricName(ctx context.Context, from, through mode
 	incomingChunkSets := make(chan ByKey)
 	incomingErrors := make(chan error)
 	for _, matcher := range matchers {
-		go func(matcher *metric.LabelMatcher) {
+		go func(matcher *labels.Matcher) {
 			// Lookup IndexQuery's
 			var queries []IndexQuery
 			var err error
-			if matcher.Type != metric.Equal {
-				queries, err = c.schema.GetReadQueriesForMetricLabel(from, through, userID, metricName, matcher.Name)
+			if matcher.Type != labels.MatchEqual {
+				queries, err = c.schema.GetReadQueriesForMetricLabel(from, through, userID, model.LabelValue(metricName), model.LabelName(matcher.Name))
 			} else {
-				queries, err = c.schema.GetReadQueriesForMetricLabelValue(from, through, userID, metricName, matcher.Name, matcher.Value)
+				queries, err = c.schema.GetReadQueriesForMetricLabelValue(from, through, userID, model.LabelValue(metricName), model.LabelName(matcher.Name), model.LabelValue(matcher.Value))
 			}
 			if err != nil {
 				incomingErrors <- err
@@ -409,7 +426,7 @@ func (c *Store) lookupEntriesByQuery(ctx context.Context, query IndexQuery) ([]I
 	return entries, nil
 }
 
-func (c *Store) convertIndexEntriesToChunks(ctx context.Context, entries []IndexEntry, matcher *metric.LabelMatcher) (ByKey, error) {
+func (c *Store) convertIndexEntriesToChunks(ctx context.Context, entries []IndexEntry, matcher *labels.Matcher) (ByKey, error) {
 	userID, err := user.ExtractOrgID(ctx)
 	if err != nil {
 		return nil, err
@@ -437,7 +454,7 @@ func (c *Store) convertIndexEntriesToChunks(ctx context.Context, entries []Index
 			chunk.metadataInIndex = true
 		}
 
-		if matcher != nil && !matcher.Match(labelValue) {
+		if matcher != nil && !matcher.Matches(string(labelValue)) {
 			util.WithContext(ctx).Debug("Dropping chunk for non-matching metric ", chunk.Metric)
 			continue
 		}