[coordinator] Allow configuration of tag validation (#2647)

src/cmd/services/m3query/config/config.go

@@ -611,6 +611,12 @@ type TagOptionsConfiguration struct {
 	// Filters are optional tag filters, removing all series with tags
 	// matching the filter from computations.
 	Filters []TagFilter `yaml:"filters"`
+
+	// AllowTagNameDuplicates allows for duplicate tags to appear on series.
+	AllowTagNameDuplicates bool `yaml:"allowTagNameDuplicates"`
+
+	// AllowTagValueEmpty allows for empty tags to appear on series.
+	AllowTagValueEmpty bool `yaml:"allowTagValueEmpty"`
 }
 
 // TagFilter is a tag filter.
@@ -665,6 +671,9 @@ func TagOptionsFromConfig(cfg TagOptionsConfiguration) (models.TagOptions, error
 		opts = opts.SetFilters(filters)
 	}
 
+	opts = opts.SetAllowTagNameDuplicates(cfg.AllowTagNameDuplicates)
+	opts = opts.SetAllowTagValueEmpty(cfg.AllowTagValueEmpty)
+
 	return opts, nil
 }
 

src/query/models/options.go

@@ -28,28 +28,34 @@ import (
 )
 
 var (
-	defaultMetricName = []byte(model.MetricNameLabel)
-	defaultBucketName = []byte("le")
+	defaultMetricName             = []byte(model.MetricNameLabel)
+	defaultBucketName             = []byte("le")
+	defaultAllowTagNameDuplicates = false
+	defaultAllowTagValueEmpty     = false
 
 	errNoName   = errors.New("metric name is missing or empty")
 	errNoBucket = errors.New("bucket name is missing or empty")
 )
 
 type tagOptions struct {
-	version    int
-	idScheme   IDSchemeType
-	bucketName []byte
-	metricName []byte
-	filters    Filters
+	version                int
+	idScheme               IDSchemeType
+	bucketName             []byte
+	metricName             []byte
+	filters                Filters
+	allowTagNameDuplicates bool
+	allowTagValueEmpty     bool
 }
 
 // NewTagOptions builds a new tag options with default values.
 func NewTagOptions() TagOptions {
 	return &tagOptions{
-		version:    0,
-		metricName: defaultMetricName,
-		bucketName: defaultBucketName,
-		idScheme:   TypeLegacy,
+		version:                0,
+		metricName:             defaultMetricName,
+		bucketName:             defaultBucketName,
+		idScheme:               TypeLegacy,
+		allowTagNameDuplicates: defaultAllowTagNameDuplicates,
+		allowTagValueEmpty:     defaultAllowTagValueEmpty,
 	}
 }
 
@@ -105,8 +111,30 @@ func (o *tagOptions) Filters() Filters {
 	return o.filters
 }
 
+func (o *tagOptions) SetAllowTagNameDuplicates(value bool) TagOptions {
+	opts := *o
+	opts.allowTagNameDuplicates = value
+	return &opts
+}
+
+func (o *tagOptions) AllowTagNameDuplicates() bool {
+	return o.allowTagNameDuplicates
+}
+
+func (o *tagOptions) SetAllowTagValueEmpty(value bool) TagOptions {
+	opts := *o
+	opts.allowTagValueEmpty = value
+	return &opts
+}
+
+func (o *tagOptions) AllowTagValueEmpty() bool {
+	return o.allowTagValueEmpty
+}
+
 func (o *tagOptions) Equals(other TagOptions) bool {
 	return o.idScheme == other.IDSchemeType() &&
 		bytes.Equal(o.metricName, other.MetricName()) &&
-		bytes.Equal(o.bucketName, other.BucketName())
+		bytes.Equal(o.bucketName, other.BucketName()) &&
+		o.allowTagNameDuplicates == other.AllowTagNameDuplicates() &&
+		o.allowTagValueEmpty == other.AllowTagValueEmpty()
 }

src/query/models/tags.go

@@ -297,13 +297,17 @@ func (t Tags) validate() error {
 			}
 		}
 	} else {
+		var (
+			allowTagNameDuplicates = t.Opts.AllowTagNameDuplicates()
+			allowTagValueEmpty     = t.Opts.AllowTagValueEmpty()
+		)
 		// Sorted alphanumerically otherwise, use bytes.Compare once for
 		// both order and unique test.
 		for i, tag := range t.Tags {
 			if len(tag.Name) == 0 {
 				return fmt.Errorf("tag name empty: index=%d", i)
 			}
-			if len(tag.Value) == 0 {
+			if !allowTagValueEmpty && len(tag.Value) == 0 {
 				return fmt.Errorf("tag value empty: index=%d, name=%s",
 					i, t.Tags[i].Name)
 			}
@@ -317,7 +321,7 @@ func (t Tags) validate() error {
 				return fmt.Errorf("tags out of order: '%s' appears after '%s', tags: %v",
 					prev.Name, tag.Name, t.Tags)
 			}
-			if cmp == 0 {
+			if !allowTagNameDuplicates && cmp == 0 {
 				return fmt.Errorf("tags duplicate: '%s' appears more than once in '%s'",
 					prev.Name, t)
 			}

src/query/models/tags_test.go

@@ -439,6 +439,15 @@ func TestTagsValidateEmptyValueQuoted(t *testing.T) {
 	require.True(t, xerrors.IsInvalidParams(err))
 }
 
+func TestTagsValidateEmptyValueQuotedWithAllowTagValueEmpty(t *testing.T) {
+	tags := NewTags(0, NewTagOptions().
+		SetIDSchemeType(TypeQuoted).
+		SetAllowTagValueEmpty(true))
+	tags = tags.AddTag(Tag{Name: []byte("foo"), Value: []byte("")})
+	err := tags.Validate()
+	require.NoError(t, err)
+}
+
 func TestTagsValidateOutOfOrderQuoted(t *testing.T) {
 	tags := NewTags(0, NewTagOptions().SetIDSchemeType(TypeQuoted))
 	tags.Tags = []Tag{
@@ -479,6 +488,26 @@ func TestTagsValidateDuplicateQuoted(t *testing.T) {
 	require.True(t, xerrors.IsInvalidParams(err))
 }
 
+func TestTagsValidateDuplicateQuotedWithAllowTagNameDuplicates(t *testing.T) {
+	tags := NewTags(0, NewTagOptions().
+		SetIDSchemeType(TypeQuoted).
+		SetAllowTagNameDuplicates(true))
+	tags = tags.AddTag(Tag{
+		Name:  []byte("foo"),
+		Value: []byte("bar"),
+	})
+	tags = tags.AddTag(Tag{
+		Name:  []byte("bar"),
+		Value: []byte("baz"),
+	})
+	tags = tags.AddTag(Tag{
+		Name:  []byte("foo"),
+		Value: []byte("qux"),
+	})
+	err := tags.Validate()
+	require.NoError(t, err)
+}
+
 func TestTagsValidateEmptyNameGraphite(t *testing.T) {
 	tags := NewTags(0, NewTagOptions().SetIDSchemeType(TypeGraphite))
 	tags = tags.AddTag(Tag{Name: nil, Value: []byte("bar")})

src/query/models/types.go

@@ -110,6 +110,18 @@ type TagOptions interface {
 	// Filters gets the tag filters.
 	Filters() Filters
 
+	// SetAllowTagNameDuplicates sets the value to allow duplicate tags to appear.
+	SetAllowTagNameDuplicates(value bool) TagOptions
+
+	// AllowTagNameDuplicates returns the value to allow duplicate tags to appear.
+	AllowTagNameDuplicates() bool
+
+	// SetAllowTagValueEmpty sets the value to allow empty tag values to appear.
+	SetAllowTagValueEmpty(value bool) TagOptions
+
+	// AllowTagValueEmpty returns the value to allow empty tag values to appear.
+	AllowTagValueEmpty() bool
+
 	// Equals determines if two tag options are equivalent.
 	Equals(other TagOptions) bool
 }