Merge "Fix memory series indexing bug."

storage/metric/memory.go

@@ -176,7 +176,7 @@ type memorySeriesStorage struct {
 
 	wmCache                 *watermarkCache
 	fingerprintToSeries     map[clientmodel.Fingerprint]stream
-	labelPairToFingerprints map[LabelPair]clientmodel.Fingerprints
+	labelPairToFingerprints map[LabelPair]utility.Set
 }
 
 // MemorySeriesOptions bundles options used by NewMemorySeriesStorage to create
@@ -240,9 +240,13 @@ func (s *memorySeriesStorage) getOrCreateSeries(metric clientmodel.Metric, finge
 				Name:  k,
 				Value: v,
 			}
-			labelPairValues := s.labelPairToFingerprints[labelPair]
-			labelPairValues = append(labelPairValues, fingerprint)
-			s.labelPairToFingerprints[labelPair] = labelPairValues
+
+			set, ok := s.labelPairToFingerprints[labelPair]
+			if !ok {
+				set = utility.Set{}
+				s.labelPairToFingerprints[labelPair] = set
+			}
+			set.Add(*fingerprint)
 		}
 	}
 	return series
@@ -267,7 +271,9 @@ func (s *memorySeriesStorage) Flush(flushOlderThan clientmodel.Timestamp, queue
 		// BUG(all): this can deadlock if the queue is full, as we only ever clear
 		// the queue after calling this method:
 		// https://github.com/prometheus/prometheus/issues/275
+		if len(queued) > 0 {
 			queue <- queued
+		}
 
 		if stream.size() == 0 {
 			emptySeries = append(emptySeries, fingerprint)
@@ -280,7 +286,6 @@ func (s *memorySeriesStorage) Flush(flushOlderThan clientmodel.Timestamp, queue
 			s.Lock()
 			s.dropSeries(&fingerprint)
 			s.Unlock()
-
 		}
 	}
 }
@@ -291,13 +296,19 @@ func (s *memorySeriesStorage) dropSeries(fingerprint *clientmodel.Fingerprint) {
 	if !ok {
 		return
 	}
+
 	for k, v := range series.metric() {
 		labelPair := LabelPair{
 			Name:  k,
 			Value: v,
 		}
+		if set, ok := s.labelPairToFingerprints[labelPair]; ok {
+			set.Remove(*fingerprint)
+			if len(set) == 0 {
 				delete(s.labelPairToFingerprints, labelPair)
 			}
+		}
+	}
 	delete(s.fingerprintToSeries, *fingerprint)
 }
 
@@ -317,32 +328,33 @@ func (s *memorySeriesStorage) appendSamplesWithoutIndexing(fingerprint *clientmo
 	series.add(samples...)
 }
 
-func (s *memorySeriesStorage) GetFingerprintsForLabelSet(l clientmodel.LabelSet) (fingerprints clientmodel.Fingerprints, err error) {
+func (s *memorySeriesStorage) GetFingerprintsForLabelSet(l clientmodel.LabelSet) (clientmodel.Fingerprints, error) {
 	s.RLock()
 	defer s.RUnlock()
 
 	sets := []utility.Set{}
 	for k, v := range l {
-		values := s.labelPairToFingerprints[LabelPair{
+		set, ok := s.labelPairToFingerprints[LabelPair{
 			Name:  k,
 			Value: v,
 		}]
-		set := utility.Set{}
-		for _, fingerprint := range values {
-			set.Add(*fingerprint)
+		if !ok {
+			return nil, nil
 		}
 		sets = append(sets, set)
 	}
 
 	setCount := len(sets)
 	if setCount == 0 {
-		return fingerprints, nil
+		return nil, nil
 	}
 
 	base := sets[0]
 	for i := 1; i < setCount; i++ {
 		base = base.Intersection(sets[i])
 	}
+
+	fingerprints := clientmodel.Fingerprints{}
 	for _, e := range base.Elements() {
 		fingerprint := e.(clientmodel.Fingerprint)
 		fingerprints = append(fingerprints, &fingerprint)
@@ -430,8 +442,8 @@ func (s *memorySeriesStorage) Close() {
 	s.Lock()
 	defer s.Unlock()
 
-	s.fingerprintToSeries = map[clientmodel.Fingerprint]stream{}
-	s.labelPairToFingerprints = map[LabelPair]clientmodel.Fingerprints{}
+	s.fingerprintToSeries = nil
+	s.labelPairToFingerprints = nil
 }
 
 func (s *memorySeriesStorage) GetAllValuesForLabel(labelName clientmodel.LabelName) (values clientmodel.LabelValues, err error) {
@@ -455,7 +467,7 @@ func (s *memorySeriesStorage) GetAllValuesForLabel(labelName clientmodel.LabelNa
 func NewMemorySeriesStorage(o MemorySeriesOptions) *memorySeriesStorage {
 	return &memorySeriesStorage{
 		fingerprintToSeries:     make(map[clientmodel.Fingerprint]stream),
-		labelPairToFingerprints: make(map[LabelPair]clientmodel.Fingerprints),
+		labelPairToFingerprints: make(map[LabelPair]utility.Set),
 		wmCache:                 o.WatermarkCache,
 	}
 }

storage/metric/memory_test.go

@@ -93,3 +93,62 @@ func BenchmarkAppendSample100(b *testing.B) {
 func BenchmarkAppendSample1000(b *testing.B) {
 	benchmarkAppendSamples(b, 1000)
 }
+
+// Regression test for https://github.com/prometheus/prometheus/issues/381.
+//
+// 1. Creates samples for two timeseries with one common labelpair.
+// 2. Flushes memory storage such that only one series is dropped from memory.
+// 3. Gets fingerprints for common labelpair.
+// 4. Checks that exactly one fingerprint remains.
+func TestDroppedSeriesIndexRegression(t *testing.T) {
+	samples := clientmodel.Samples{
+		&clientmodel.Sample{
+			Metric: clientmodel.Metric{
+				clientmodel.MetricNameLabel: "testmetric",
+				"different":                 "differentvalue1",
+				"common":                    "samevalue",
+			},
+			Value:     1,
+			Timestamp: clientmodel.TimestampFromTime(time.Date(2000, 0, 0, 0, 0, 0, 0, time.UTC)),
+		},
+		&clientmodel.Sample{
+			Metric: clientmodel.Metric{
+				clientmodel.MetricNameLabel: "testmetric",
+				"different":                 "differentvalue2",
+				"common":                    "samevalue",
+			},
+			Value:     2,
+			Timestamp: clientmodel.TimestampFromTime(time.Date(2002, 0, 0, 0, 0, 0, 0, time.UTC)),
+		},
+	}
+
+	s := NewMemorySeriesStorage(MemorySeriesOptions{})
+	s.AppendSamples(samples)
+
+	common := clientmodel.LabelSet{"common": "samevalue"}
+	fps, err := s.GetFingerprintsForLabelSet(common)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(fps) != 2 {
+		t.Fatalf("Got %d fingerprints, expected 2", len(fps))
+	}
+
+	toDisk := make(chan clientmodel.Samples, 2)
+	s.Flush(clientmodel.TimestampFromTime(time.Date(2001, 0, 0, 0, 0, 0, 0, time.UTC)), toDisk)
+	if len(toDisk) != 1 {
+		t.Fatalf("Got %d disk sample lists, expected 1", len(toDisk))
+	}
+	diskSamples := <-toDisk
+	if len(diskSamples) != 1 {
+		t.Fatalf("Got %d disk samples, expected 1", len(diskSamples))
+	}
+
+	fps, err = s.GetFingerprintsForLabelSet(common)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(fps) != 1 {
+		t.Fatalf("Got %d fingerprints, expected 1", len(fps))
+	}
+}