feat(filter): loose tag regex and compatibility (#924)

by @lordvidex

cmd/filter/compatibility.go

@@ -0,0 +1,25 @@
+package main
+
+import (
+	"github.com/moira-alert/moira/filter"
+)
+
+// Compatibility struct contains feature-flags that give user control over
+// features supported by different versions of graphit compatible with moira
+type compatibility struct {
+	// Controls how regices in tag matching are treated
+	// If false (default value), regex will match start of the string strictly. 'tag~=foo' is equivalent to 'tag~=^foo.*'
+	// If true, regex will match start of the string loosely. 'tag~=foo' is equivalent to 'tag~=.*foo.*'
+	AllowRegexLooseStartMatch bool `yaml:"allow_regex_loose_start_match"`
+	// Controls how absent tags are treated
+	// If true (default value), empty tags in regices will be matched
+	// If false, empty tags will be discarded
+	AllowRegexMatchEmpty bool `yaml:"allow_regex_match_empty"`
+}
+
+func (compatibility *compatibility) toFilterCompatibility() filter.Compatibility {
+	return filter.Compatibility{
+		AllowRegexLooseStartMatch: compatibility.AllowRegexLooseStartMatch,
+		AllowRegexMatchEmpty:      compatibility.AllowRegexMatchEmpty,
+	}
+}

cmd/filter/config.go

@@ -28,6 +28,8 @@ type filterConfig struct {
 	PatternsUpdatePeriod string `yaml:"patterns_update_period"`
 	// DropMetricsTTL this is time window how older metric we can get from now.
 	DropMetricsTTL string `yaml:"drop_metrics_ttl"`
+	// Flags for compatibility with different graphite behaviours
+	Compatibility compatibility `yaml:"graphite_compatibility"`
 }
 
 func getDefault() config {
@@ -49,6 +51,10 @@ func getDefault() config {
 			MaxParallelMatches:   0,
 			PatternsUpdatePeriod: "1s",
 			DropMetricsTTL:       "1h",
+			Compatibility: compatibility{
+				AllowRegexLooseStartMatch: false,
+				AllowRegexMatchEmpty:      true,
+			},
 		},
 		Telemetry: cmd.TelemetryConfig{
 			Listen: ":8094",

cmd/filter/main.go

@@ -69,6 +69,8 @@ func main() {
 		String("moira_version", MoiraVersion).
 		Msg("Moira Filter stopped. Version")
 
+	compatibility := config.Filter.Compatibility.toFilterCompatibility()
+
 	telemetry, err := cmd.ConfigureTelemetry(logger, config.Telemetry, serviceName)
 	if err != nil {
 		logger.Fatal().
@@ -103,7 +105,7 @@ func main() {
 			Msg("Failed to initialize cache storage with given config")
 	}
 
-	patternStorage, err := filter.NewPatternStorage(database, filterMetrics, logger)
+	patternStorage, err := filter.NewPatternStorage(database, filterMetrics, logger, compatibility)
 	if err != nil {
 		logger.Fatal().
 			Error(err).

filter/compatibility.go

@@ -0,0 +1,7 @@
+package filter
+
+// See cmd/filter/compatibility for usage examples
+type Compatibility struct {
+	AllowRegexLooseStartMatch bool
+	AllowRegexMatchEmpty      bool
+}

filter/pattern_index.go

@@ -6,17 +6,18 @@ import (
 
 // PatternIndex helps to index patterns and allows to match them by metric
 type PatternIndex struct {
-	Tree *PrefixTree
+	Tree          *PrefixTree
+	compatibility Compatibility
 }
 
 // NewPatternIndex creates new PatternIndex using patterns
-func NewPatternIndex(logger moira.Logger, patterns []string) *PatternIndex {
+func NewPatternIndex(logger moira.Logger, patterns []string, compatibility Compatibility) *PatternIndex {
 	prefixTree := &PrefixTree{Logger: logger, Root: &PatternNode{}}
 	for _, pattern := range patterns {
 		prefixTree.Add(pattern)
 	}
 
-	return &PatternIndex{Tree: prefixTree}
+	return &PatternIndex{Tree: prefixTree, compatibility: compatibility}
 }
 
 // MatchPatterns allows matching pattern by metric

filter/pattern_index_test.go

@@ -26,7 +26,7 @@ func TestPatternIndex(t *testing.T) {
 			"Question.at_the_end?",
 		}
 
-		index := NewPatternIndex(logger, patterns)
+		index := NewPatternIndex(logger, patterns, Compatibility{AllowRegexLooseStartMatch: true})
 		testCases := []struct {
 			Metric          string
 			MatchedPatterns []string

filter/patterns_storage.go

@@ -19,15 +19,22 @@ type PatternStorage struct {
 	logger                  moira.Logger
 	PatternIndex            atomic.Value
 	SeriesByTagPatternIndex atomic.Value
+	compatibility           Compatibility
 }
 
 // NewPatternStorage creates new PatternStorage struct
-func NewPatternStorage(database moira.Database, metrics *metrics.FilterMetrics, logger moira.Logger) (*PatternStorage, error) {
+func NewPatternStorage(
+	database moira.Database,
+	metrics *metrics.FilterMetrics,
+	logger moira.Logger,
+	compatibility Compatibility,
+) (*PatternStorage, error) {
 	storage := &PatternStorage{
-		database: database,
-		metrics:  metrics,
-		logger:   logger,
-		clock:    clock.NewSystemClock(),
+		database:      database,
+		metrics:       metrics,
+		logger:        logger,
+		clock:         clock.NewSystemClock(),
+		compatibility: compatibility,
 	}
 	err := storage.Refresh()
 	return storage, err
@@ -51,8 +58,8 @@ func (storage *PatternStorage) Refresh() error {
 		}
 	}
 
-	storage.PatternIndex.Store(NewPatternIndex(storage.logger, patterns))
-	storage.SeriesByTagPatternIndex.Store(NewSeriesByTagPatternIndex(storage.logger, seriesByTagPatterns))
+	storage.PatternIndex.Store(NewPatternIndex(storage.logger, patterns, storage.compatibility))
+	storage.SeriesByTagPatternIndex.Store(NewSeriesByTagPatternIndex(storage.logger, seriesByTagPatterns, storage.compatibility))
 	return nil
 }
 

filter/patterns_storage_test.go

@@ -30,12 +30,17 @@ func TestProcessIncomingMetric(t *testing.T) {
 	Convey("Create new pattern storage, GetPatterns returns error, should error", t, func() {
 		database.EXPECT().GetPatterns().Return(nil, fmt.Errorf("some error here"))
 		filterMetrics := metrics.ConfigureFilterMetrics(metrics.NewDummyRegistry())
-		_, err := NewPatternStorage(database, filterMetrics, logger)
+		_, err := NewPatternStorage(database, filterMetrics, logger, Compatibility{AllowRegexLooseStartMatch: true})
 		So(err, ShouldBeError, fmt.Errorf("some error here"))
 	})
 
 	database.EXPECT().GetPatterns().Return(testPatterns, nil)
-	patternsStorage, err := NewPatternStorage(database, metrics.ConfigureFilterMetrics(metrics.NewDummyRegistry()), logger)
+	patternsStorage, err := NewPatternStorage(
+		database,
+		metrics.ConfigureFilterMetrics(metrics.NewDummyRegistry()),
+		logger,
+		Compatibility{AllowRegexLooseStartMatch: true},
+	)
 	systemClock := mock_clock.NewMockClock(mockCtrl)
 	systemClock.EXPECT().Now().Return(time.Date(2009, 2, 13, 23, 31, 30, 0, time.UTC)).AnyTimes()
 	patternsStorage.clock = systemClock

filter/series_by_tag.go

@@ -129,15 +129,16 @@ func ParseSeriesByTag(input string) ([]TagSpec, error) {
 type MatchingHandler func(string, map[string]string) bool
 
 // CreateMatchingHandlerForPattern creates function for matching by tag list
-func CreateMatchingHandlerForPattern(tagSpecs []TagSpec) (string, MatchingHandler) {
+func CreateMatchingHandlerForPattern(tagSpecs []TagSpec, compatibility *Compatibility) (string, MatchingHandler) {
 	matchingHandlers := make([]MatchingHandler, 0)
 	var nameTagValue string
 
 	for _, tagSpec := range tagSpecs {
 		if tagSpec.Name == "name" && tagSpec.Operator == EqualOperator {
 			nameTagValue = tagSpec.Value
 		} else {
-			matchingHandlers = append(matchingHandlers, createMatchingHandlerForOneTag(tagSpec))
+			handler := createMatchingHandlerForOneTag(tagSpec, compatibility)
+			matchingHandlers = append(matchingHandlers, handler)
 		}
 	}
 
@@ -153,9 +154,10 @@ func CreateMatchingHandlerForPattern(tagSpecs []TagSpec) (string, MatchingHandle
 	return nameTagValue, matchingHandler
 }
 
-func createMatchingHandlerForOneTag(spec TagSpec) MatchingHandler {
+func createMatchingHandlerForOneTag(spec TagSpec, compatibility *Compatibility) MatchingHandler {
 	var matchingHandlerCondition func(string) bool
 	allowMatchEmpty := false
+
 	switch spec.Operator {
 	case EqualOperator:
 		allowMatchEmpty = true
@@ -167,13 +169,18 @@ func createMatchingHandlerForOneTag(spec TagSpec) MatchingHandler {
 			return value != spec.Value
 		}
 	case MatchOperator:
-		allowMatchEmpty = true
-		matchRegex := regexp.MustCompile("^" + spec.Value)
+		allowMatchEmpty = compatibility.AllowRegexMatchEmpty
+
+		matchRegex := newMatchRegex(spec.Value, compatibility)
+
 		matchingHandlerCondition = func(value string) bool {
 			return matchRegex.MatchString(value)
 		}
 	case NotMatchOperator:
-		matchRegex := regexp.MustCompile("^" + spec.Value)
+		allowMatchEmpty = compatibility.AllowRegexMatchEmpty
+
+		matchRegex := newMatchRegex(spec.Value, compatibility)
+
 		matchingHandlerCondition = func(value string) bool {
 			return !matchRegex.MatchString(value)
 		}
@@ -194,3 +201,19 @@ func createMatchingHandlerForOneTag(spec TagSpec) MatchingHandler {
 		return allowMatchEmpty && matchEmpty
 	}
 }
+
+func newMatchRegex(value string, compatibility *Compatibility) *regexp.Regexp {
+	var matchRegex *regexp.Regexp
+
+	if value == "*" {
+		value = ".*"
+	}
+
+	if compatibility.AllowRegexLooseStartMatch {
+		matchRegex = regexp.MustCompile(value)
+	} else {
+		matchRegex = regexp.MustCompile("^" + value)
+	}
+
+	return matchRegex
+}

filter/series_by_tag_pattern_index.go

@@ -1,22 +1,30 @@
 package filter
 
-import "github.com/moira-alert/moira"
+import (
+	"github.com/moira-alert/moira"
+)
 
 // SeriesByTagPatternIndex helps to index the seriesByTag patterns and allows to match them by metric
 type SeriesByTagPatternIndex struct {
 	// namesPrefixTree stores MatchingHandler's for patterns that have name tag in prefix tree structure
 	namesPrefixTree *PrefixTree
 	// withoutStrictNameTagPatternMatchers stores MatchingHandler's for patterns that have no name tag
 	withoutStrictNameTagPatternMatchers map[string]MatchingHandler
+	// Flags for compatibility with different graphite behaviours
+	compatibility Compatibility
 }
 
 // NewSeriesByTagPatternIndex creates new SeriesByTagPatternIndex using seriesByTag patterns and parsed specs comes from ParseSeriesByTag
-func NewSeriesByTagPatternIndex(logger moira.Logger, tagSpecsByPattern map[string][]TagSpec) *SeriesByTagPatternIndex {
+func NewSeriesByTagPatternIndex(
+	logger moira.Logger,
+	tagSpecsByPattern map[string][]TagSpec,
+	compatibility Compatibility,
+) *SeriesByTagPatternIndex {
 	namesPrefixTree := &PrefixTree{Logger: logger, Root: &PatternNode{}}
 	withoutStrictNameTagPatternMatchers := make(map[string]MatchingHandler)
 
 	for pattern, tagSpecs := range tagSpecsByPattern {
-		nameTagValue, matchingHandler := CreateMatchingHandlerForPattern(tagSpecs)
+		nameTagValue, matchingHandler := CreateMatchingHandlerForPattern(tagSpecs, &compatibility)
 
 		if nameTagValue == "" {
 			withoutStrictNameTagPatternMatchers[pattern] = matchingHandler
@@ -26,8 +34,10 @@ func NewSeriesByTagPatternIndex(logger moira.Logger, tagSpecsByPattern map[strin
 	}
 
 	return &SeriesByTagPatternIndex{
+		compatibility:                       compatibility,
 		namesPrefixTree:                     namesPrefixTree,
-		withoutStrictNameTagPatternMatchers: withoutStrictNameTagPatternMatchers}
+		withoutStrictNameTagPatternMatchers: withoutStrictNameTagPatternMatchers,
+	}
 }
 
 // MatchPatterns allows to match patterns by metric name and its labels
@@ -42,7 +52,7 @@ func (index *SeriesByTagPatternIndex) MatchPatterns(metricName string, labels ma
 	}
 
 	for pattern, matchingHandler := range index.withoutStrictNameTagPatternMatchers {
-		if (matchingHandler)(metricName, labels) {
+		if matchingHandler(metricName, labels) {
 			matchedPatterns = append(matchedPatterns, pattern)
 		}
 	}