Merge branch 'fixfilter'

cmd/filter/main.go

@@ -107,7 +107,7 @@ func main() {
 	defer stopHeartbeatWorker(heartbeatWorker)
 
 	// Start metrics listener
-	listener, err := connection.NewListener(config.Filter.Listen, logger, patternStorage)
+	listener, err := connection.NewListener(config.Filter.Listen, logger, cacheMetrics, patternStorage)
 	if err != nil {
 		logger.Fatalf("Failed to start listen: %s", err.Error())
 	}

database/redis/database.go

@@ -28,24 +28,26 @@ const (
 
 // DbConnector contains redis pool
 type DbConnector struct {
-	pool            *redis.Pool
-	logger          moira.Logger
-	retentionCache  *cache.Cache
-	metricsCache    *cache.Cache
-	messengersCache *cache.Cache
-	sync            *redsync.Redsync
+	pool                 *redis.Pool
+	logger               moira.Logger
+	retentionCache       *cache.Cache
+	retentionSavingCache *cache.Cache
+	metricsCache         *cache.Cache
+	messengersCache      *cache.Cache
+	sync                 *redsync.Redsync
 }
 
 // NewDatabase creates Redis pool based on config
 func NewDatabase(logger moira.Logger, config Config) *DbConnector {
 	pool := newRedisPool(logger, config)
 	return &DbConnector{
-		pool:            pool,
-		logger:          logger,
-		retentionCache:  cache.New(cacheValueExpirationDuration, cacheCleanupInterval),
-		metricsCache:    cache.New(cacheValueExpirationDuration, cacheCleanupInterval),
-		messengersCache: cache.New(cache.NoExpiration, cache.DefaultExpiration),
-		sync:            redsync.New([]redsync.Pool{pool}),
+		pool:                 pool,
+		logger:               logger,
+		retentionCache:       cache.New(cacheValueExpirationDuration, cacheCleanupInterval),
+		retentionSavingCache: cache.New(cache.NoExpiration, cache.DefaultExpiration),
+		metricsCache:         cache.New(cacheValueExpirationDuration, cacheCleanupInterval),
+		messengersCache:      cache.New(cache.NoExpiration, cache.DefaultExpiration),
+		sync:                 redsync.New([]redsync.Pool{pool}),
 	}
 }
 

database/redis/metric.go

@@ -10,6 +10,7 @@ import (
 	"github.com/moira-alert/moira"
 	"github.com/moira-alert/moira/database"
 	"github.com/moira-alert/moira/database/redis/reply"
+	"github.com/patrickmn/go-cache"
 )
 
 // GetPatterns gets updated patterns array
@@ -96,7 +97,10 @@ func (connector *DbConnector) SaveMetrics(metrics map[string]*moira.MatchedMetri
 	for _, metric := range metrics {
 		metricValue := fmt.Sprintf("%v %v", metric.Timestamp, metric.Value)
 		c.Send("ZADD", metricDataKey(metric.Metric), metric.RetentionTimestamp, metricValue)
-		c.Send("SET", metricRetentionKey(metric.Metric), metric.Retention)
+
+		if err := connector.retentionSavingCache.Add(metric.Metric, true, cache.DefaultExpiration); err == nil {
+			c.Send("SET", metricRetentionKey(metric.Metric), metric.Retention)
+		}
 
 		for _, pattern := range metric.Patterns {
 			c.Send("SADD", patternMetricsKey(pattern), metric.Metric)

filter/connection/listening.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/moira-alert/moira"
 	"github.com/moira-alert/moira/filter"
+	"github.com/moira-alert/moira/metrics/graphite"
 )
 
 // MetricsListener is facade for standard net.MetricsListener and accept connection for handling it
@@ -17,10 +18,11 @@ type MetricsListener struct {
 	handler  *Handler
 	logger   moira.Logger
 	tomb     tomb.Tomb
+	metrics  *graphite.FilterMetrics
 }
 
 // NewListener creates new listener
-func NewListener(port string, logger moira.Logger, patternStorage *filter.PatternStorage) (*MetricsListener, error) {
+func NewListener(port string, logger moira.Logger, metrics *graphite.FilterMetrics, patternStorage *filter.PatternStorage) (*MetricsListener, error) {
 	address, err := net.ResolveTCPAddr("tcp", port)
 	if nil != err {
 		return nil, fmt.Errorf("Failed to resolve tcp address [%s]: %s", port, err.Error())
@@ -33,14 +35,15 @@ func NewListener(port string, logger moira.Logger, patternStorage *filter.Patter
 		listener: newListener,
 		logger:   logger,
 		handler:  NewConnectionsHandler(logger, patternStorage),
+		metrics: 	metrics,
 	}
 	return &listener, nil
 }
 
 // Listen waits for new data in connection and handles it in ConnectionHandler
 // All handled data sets to metricsChan
 func (listener *MetricsListener) Listen() chan *moira.MatchedMetric {
-	metricsChan := make(chan *moira.MatchedMetric, 10000)
+	metricsChan := make(chan *moira.MatchedMetric, 16384)
 	listener.tomb.Go(func() error {
 		for {
 			select {
@@ -68,10 +71,24 @@ func (listener *MetricsListener) Listen() chan *moira.MatchedMetric {
 			listener.handler.HandleConnection(conn, metricsChan)
 		}
 	})
+
+	listener.tomb.Go(func() error { return listener.checkNewMetricsChannelLen(metricsChan) })
 	listener.logger.Info("Moira Filter Listener Started")
 	return metricsChan
 }
 
+func (listener *MetricsListener) checkNewMetricsChannelLen(channel <-chan *moira.MatchedMetric) error {
+	checkTicker := time.NewTicker(time.Millisecond * 100)
+	for {
+		select {
+		case <-listener.tomb.Dying():
+			return nil
+		case <-checkTicker.C:
+			listener.metrics.MetricChannelLen.Update(int64(len(channel)))
+		}
+	}
+}
+
 // Stop stops listening connection
 func (listener *MetricsListener) Stop() error {
 	listener.tomb.Kill(nil)

filter/matched_metrics/metrics.go

@@ -43,7 +43,7 @@ func (matcher *MetricsMatcher) Start(channel chan *moira.MatchedMetric) {
 					return
 				}
 				matcher.cacheStorage.EnrichMatchedMetric(buffer, metric)
-				if len(buffer) < 10 {
+				if len(buffer) < 100 {
 					continue
 				}
 			case <-time.After(time.Second):

metrics/graphite/filter.go

@@ -8,4 +8,5 @@ type FilterMetrics struct {
 	MatchingTimer           Timer
 	SavingTimer             Timer
 	BuildTreeTimer          Timer
+	MetricChannelLen        Histogram
 }

metrics/graphite/go-metrics/configuration.go

@@ -15,6 +15,7 @@ func ConfigureFilterMetrics(prefix string) *graphite.FilterMetrics {
 		MatchingTimer:           registerTimer(metricNameWithPrefix(prefix, "time.match")),
 		SavingTimer:             registerTimer(metricNameWithPrefix(prefix, "time.save")),
 		BuildTreeTimer:          registerTimer(metricNameWithPrefix(prefix, "time.buildtree")),
+		MetricChannelLen:        registerHistogram(metricNameWithPrefix(prefix, "metricsToSave")),
 	}
 }