moira

1
package filter
2

3
import (
4
	"bufio"
5
	"io"
6
	"regexp"
7
	"strconv"
8
	"strings"
9

10
	"go.avito.ru/DO/moira"
11
)
12

13
const defaultRetention = 60
14

15
type retentionMatcher struct {
16
	pattern   *regexp.Regexp
17
	retention int
18
}
19

20
type retentionCacheItem struct {
21
	value     int
22
	timestamp int64
23
}
24

25
// Storage struct to store retention matchers
26
type Storage struct {
27
	retentions      []retentionMatcher
28
	retentionsCache map[string]*retentionCacheItem
29
	metricsCache    map[string]*moira.MatchedMetric
30
}
31

32
// NewCacheStorage create new Storage
33
func NewCacheStorage(reader io.Reader) (*Storage, error) {
34
	storage := &Storage{
35
		retentionsCache: make(map[string]*retentionCacheItem),
36
		metricsCache:    make(map[string]*moira.MatchedMetric),
37
	}
38

39
	if err := storage.buildRetentions(bufio.NewScanner(reader)); err != nil {
40
		return nil, err
41
	}
42
	return storage, nil
43
}
44

45
// EnrichMatchedMetric calculate retention and filter cached values
46
func (storage *Storage) EnrichMatchedMetric(buffer map[string]*moira.MatchedMetric, m *moira.MatchedMetric) {
47
	m.Retention = storage.getRetention(m)
48
	m.RetentionTimestamp = roundToNearestRetention(m.Timestamp, int64(m.Retention))
49
	if ex, ok := storage.metricsCache[m.Metric]; ok && ex.RetentionTimestamp == m.RetentionTimestamp && ex.Value == m.Value {
50
		return
51
	}
52
	storage.metricsCache[m.Metric] = m
53
	buffer[m.Metric] = m
54
}
55

56
// getRetention returns first matched retention for metric
57
func (storage *Storage) getRetention(m *moira.MatchedMetric) int {
58
	if item, ok := storage.retentionsCache[m.Metric]; ok && item.timestamp+60 > m.Timestamp {
59
		return item.value
60
	}
61
	for _, matcher := range storage.retentions {
62
		if matcher.pattern.MatchString(m.Metric) {
63
			storage.retentionsCache[m.Metric] = &retentionCacheItem{
64
				value:     matcher.retention,
65
				timestamp: m.Timestamp,
66
			}
67
			return matcher.retention
68
		}
69
	}
70
	return defaultRetention
71
}
72

73
func (storage *Storage) buildRetentions(retentionScanner *bufio.Scanner) error {
74
	storage.retentions = make([]retentionMatcher, 0, 100)
75

76
	for retentionScanner.Scan() {
77
		line := retentionScanner.Text()
78
		if strings.HasPrefix(line, "#") || strings.Count(line, "=") != 1 {
79
			continue
80
		}
81

82
		pattern, err := regexp.Compile(strings.TrimSpace(strings.Split(line, "=")[1]))
83
		if err != nil {
84
			return err
85
		}
86

87
		retentionScanner.Scan()
88
		line = retentionScanner.Text()
89
		retentions := strings.TrimSpace(strings.Split(line, "=")[1])
90
		retention, err := rawRetentionToSeconds(retentions[0:strings.Index(retentions, ":")])
91
		if err != nil {
92
			return err
93
		}
94

95
		storage.retentions = append(storage.retentions, retentionMatcher{
96
			pattern:   pattern,
97
			retention: retention,
98
		})
99
	}
100
	return retentionScanner.Err()
101
}
102

103
func rawRetentionToSeconds(rawRetention string) (int, error) {
104
	retention, err := strconv.Atoi(rawRetention)
105
	if err == nil {
106
		return retention, nil
107
	}
108

109
	multiplier := 1
110
	switch {
111
	case strings.HasSuffix(rawRetention, "m"):
112
		multiplier = 60
113
	case strings.HasSuffix(rawRetention, "h"):
114
		multiplier = 60 * 60
115
	case strings.HasSuffix(rawRetention, "d"):
116
		multiplier = 60 * 60 * 24
117
	case strings.HasSuffix(rawRetention, "w"):
118
		multiplier = 60 * 60 * 24 * 7
119
	case strings.HasSuffix(rawRetention, "y"):
120
		multiplier = 60 * 60 * 24 * 365
121
	}
122

123
	retention, err = strconv.Atoi(rawRetention[0 : len(rawRetention)-1])
124
	if err != nil {
125
		return 0, err
126
	}
127

128
	return retention * multiplier, nil
129
}
130

131
func roundToNearestRetention(ts, retention int64) int64 {
132
	return (ts + retention/2) / retention * retention
133
}
134