Correctly handle empty labels. (#594)

Currently a time series with empty labels is not treated the same
as one with missing labels. Currently this can only come from
ALERTS&ALERT_FOR_STATE so it's unlikely anyone has actually hit it.

Signed-off-by: Brian Brazil <brian.brazil@robustperception.io>

db_test.go

@@ -219,6 +219,29 @@ func TestDBAppenderAddRef(t *testing.T) {
 	}, res)
 }
 
+func TestAppendEmptyLabelsIgnored(t *testing.T) {
+	db, delete := openTestDB(t, nil)
+	defer func() {
+		testutil.Ok(t, db.Close())
+		delete()
+	}()
+
+	app1 := db.Appender()
+
+	ref1, err := app1.Add(labels.FromStrings("a", "b"), 123, 0)
+	testutil.Ok(t, err)
+
+	// Construct labels manually so there is an empty label.
+	ref2, err := app1.Add(labels.Labels{labels.Label{"a", "b"}, labels.Label{"c", ""}}, 124, 0)
+	testutil.Ok(t, err)
+
+	// Should be the same series.
+	testutil.Equals(t, ref1, ref2)
+
+	err = app1.Commit()
+	testutil.Ok(t, err)
+}
+
 func TestDeleteSimple(t *testing.T) {
 	numSamples := int64(10)
 
@@ -1580,26 +1603,26 @@ func TestDB_LabelNames(t *testing.T) {
 	}{
 		{
 			sampleLabels1: [][2]string{
-				[2]string{"name1", ""},
+				[2]string{"name1", "1"},
-				[2]string{"name3", ""},
+				[2]string{"name3", "3"},
-				[2]string{"name2", ""},
+				[2]string{"name2", "2"},
 			},
 			sampleLabels2: [][2]string{
-				[2]string{"name4", ""},
+				[2]string{"name4", "4"},
-				[2]string{"name1", ""},
+				[2]string{"name1", "1"},
 			},
 			exp1: []string{"name1", "name2", "name3"},
 			exp2: []string{"name1", "name2", "name3", "name4"},
 		},
 		{
 			sampleLabels1: [][2]string{
-				[2]string{"name2", ""},
+				[2]string{"name2", "2"},
-				[2]string{"name1", ""},
+				[2]string{"name1", "1"},
-				[2]string{"name2", ""},
+				[2]string{"name2", "2"},
 			},
 			sampleLabels2: [][2]string{
-				[2]string{"name6", ""},
+				[2]string{"name6", "6"},
-				[2]string{"name0", ""},
+				[2]string{"name0", "0"},
 			},
 			exp1: []string{"name1", "name2"},
 			exp2: []string{"name0", "name1", "name2", "name6"},

head.go

@@ -755,6 +755,9 @@ func (a *headAppender) Add(lset labels.Labels, t int64, v float64) (uint64, erro
 		return 0, ErrOutOfBounds
 	}
 
+	// Ensure no empty labels have gotten through.
+	lset = lset.WithoutEmpty()
+
 	s, created := a.head.getOrCreate(lset.Hash(), lset)
 	if created {
 		a.series = append(a.series, RefSeries{

labels/labels.go

@@ -103,6 +103,23 @@ func (ls Labels) Map() map[string]string {
 	return m
 }
 
+// WithoutEmpty returns the labelset without empty labels.
+// May return the same labelset.
+func (ls Labels) WithoutEmpty() Labels {
+	for _, v := range ls {
+		if v.Value == "" {
+			els := make(Labels, 0, len(ls)-1)
+			for _, v := range ls {
+				if v.Value != "" {
+					els = append(els, v)
+				}
+			}
+			return els
+		}
+	}
+	return ls
+}
+
 // New returns a sorted Labels from the given labels.
 // The caller has to guarantee that all label names are unique.
 func New(ls ...Label) Labels {
@@ -119,7 +136,9 @@ func New(ls ...Label) Labels {
 func FromMap(m map[string]string) Labels {
 	l := make(Labels, 0, len(m))
 	for k, v := range m {
-		l = append(l, Label{Name: k, Value: v})
+		if v != "" {
+			l = append(l, Label{Name: k, Value: v})
+		}
 	}
 	sort.Sort(l)
 
@@ -133,7 +152,9 @@ func FromStrings(ss ...string) Labels {
 	}
 	var res Labels
 	for i := 0; i < len(ss); i += 2 {
-		res = append(res, Label{Name: ss[i], Value: ss[i+1]})
+		if ss[i+1] != "" {
+			res = append(res, Label{Name: ss[i], Value: ss[i+1]})
+		}
 	}
 
 	sort.Sort(res)