Merge pull request #268 from prometheus/fix/value-extraction

Interval value extraction fixes and optimizations.

storage/metric/memory.go

@@ -181,21 +181,23 @@ func (s *memorySeriesStorage) AppendSample(sample model.Sample) error {
 	return nil
 }
 
-// Append raw sample, bypassing indexing. Only used to add data to views, which
-// don't need to lookup by metric.
-func (s *memorySeriesStorage) appendSampleWithoutIndexing(f *model.Fingerprint, timestamp time.Time, value model.SampleValue) {
+// Append raw samples, bypassing indexing. Only used to add data to views,
+// which don't need to lookup by metric.
+func (s *memorySeriesStorage) appendSamplesWithoutIndexing(fingerprint *model.Fingerprint, samples model.Values) {
 	s.RLock()
-	series, ok := s.fingerprintToSeries[*f]
+	series, ok := s.fingerprintToSeries[*fingerprint]
 	s.RUnlock()
 
 	if !ok {
 		series = newStream(model.Metric{})
 		s.Lock()
-		s.fingerprintToSeries[*f] = series
+		s.fingerprintToSeries[*fingerprint] = series
 		s.Unlock()
 	}
 
-	series.add(timestamp, value)
+	for _, sample := range samples {
+		series.add(sample.Timestamp, sample.Value)
+	}
 }
 
 func (s *memorySeriesStorage) GetFingerprintsForLabelSet(l model.LabelSet) (fingerprints model.Fingerprints, err error) {

storage/metric/operation.go

@@ -111,14 +111,14 @@ func extractValuesAroundTime(t time.Time, in model.Values) (out model.Values) {
 		if in[i].Timestamp.Equal(t) && len(in) > i+1 {
 			// We hit exactly the current sample time. Very unlikely in practice.
 			// Return only the current sample.
-			out = append(out, in[i])
+			out = in[i : i+1]
 		} else {
 			if i == 0 {
 				// We hit before the first sample time. Return only the first sample.
-				out = append(out, in[0:1]...)
+				out = in[0:1]
 			} else {
 				// We hit between two samples. Return both surrounding samples.
-				out = append(out, in[i-1:i+1]...)
+				out = in[i-1 : i+1]
 			}
 		}
 	}
@@ -161,12 +161,12 @@ func (g *getValuesAtIntervalOp) ExtractSamples(in model.Values) (out model.Value
 		lastExtractedTime := out[len(out)-1].Timestamp
 		in = in.TruncateBefore(lastExtractedTime.Add(1))
 		g.from = g.from.Add(g.interval)
-		if lastExtractedTime.Equal(lastChunkTime) {
-			break
-		}
 		for !g.from.After(lastExtractedTime) {
 			g.from = g.from.Add(g.interval)
 		}
+		if lastExtractedTime.Equal(lastChunkTime) {
+			break
+		}
 		if g.from.After(g.through) {
 			break
 		}

storage/metric/operation_test.go

@@ -1610,6 +1610,16 @@ func TestGetValuesAtIntervalOp(t *testing.T) {
 			t.Fatalf("%d. expected length %d, got %d: %v", i, len(scenario.out), len(actual), scenario.op)
 			t.Fatalf("%d. expected length %d, got %d", i, len(scenario.out), len(actual))
 		}
+
+		if len(scenario.in) < 1 {
+			continue
+		}
+		opTime := scenario.op.CurrentTime()
+		lastExtractedTime := scenario.out[len(scenario.out)-1].Timestamp
+		if opTime != nil && opTime.Before(lastExtractedTime) {
+			t.Fatalf("%d. expected op.CurrentTime() to be nil or after current chunk, %v, %v", i, scenario.op.CurrentTime(), scenario.out)
+		}
+
 		for j, out := range scenario.out {
 			if out != actual[j] {
 				t.Fatalf("%d. expected output %v, got %v", i, scenario.out, actual)

storage/metric/tiered.go

@@ -348,12 +348,10 @@ func (t *TieredStorage) renderView(viewJob viewJob) {
 
 				for op.CurrentTime() != nil && !op.CurrentTime().After(targetTime) {
 					out = op.ExtractSamples(model.Values(currentChunk))
-				}
-			}
 
-			// Append the extracted samples to the materialized view.
-			for _, sample := range out {
-				view.appendSample(scanJob.fingerprint, sample.Timestamp, sample.Value)
+					// Append the extracted samples to the materialized view.
+					view.appendSamples(scanJob.fingerprint, out)
+				}
 			}
 
 			// Throw away standing ops which are finished.

storage/metric/view.go

@@ -105,8 +105,8 @@ type view struct {
 	*memorySeriesStorage
 }
 
-func (v view) appendSample(fingerprint *model.Fingerprint, timestamp time.Time, value model.SampleValue) {
-	v.memorySeriesStorage.appendSampleWithoutIndexing(fingerprint, timestamp, value)
+func (v view) appendSamples(fingerprint *model.Fingerprint, samples model.Values) {
+	v.memorySeriesStorage.appendSamplesWithoutIndexing(fingerprint, samples)
 }
 
 func newView() view {