feat(notifier): fix sending unwanted notifications on muted and deleted triggers and muted metrics

cmd/api/main.go

@@ -83,7 +83,7 @@ func main() {
 
 	databaseSettings := applicationConfig.Redis.GetSettings()
 	notificationHistorySettings := applicationConfig.NotificationHistory.GetSettings()
-	database := redis.NewDatabase(logger, databaseSettings, notificationHistorySettings, redis.API)
+	database := redis.NewDatabase(logger, databaseSettings, notificationHistorySettings, redis.NotificationConfig{}, redis.API)
 
 	// Start Index right before HTTP listener. Fail if index cannot start
 	searchIndex := index.NewSearchIndex(logger, database, telemetry.Metrics)

cmd/checker/main.go

@@ -83,7 +83,7 @@ func main() {
 	logger.Info().Msg("Debug: checker started")
 
 	databaseSettings := config.Redis.GetSettings()
-	database := redis.NewDatabase(logger, databaseSettings, redis.NotificationHistoryConfig{}, redis.Checker)
+	database := redis.NewDatabase(logger, databaseSettings, redis.NotificationHistoryConfig{}, redis.NotificationConfig{}, redis.Checker)
 
 	remoteConfig := config.Remote.GetRemoteSourceSettings()
 	prometheusConfig := config.Prometheus.GetPrometheusSourceSettings()

cmd/cli/main.go

@@ -345,7 +345,7 @@ func initApp() (cleanupConfig, moira.Logger, moira.Database) {
 	}
 
 	databaseSettings := config.Redis.GetSettings()
-	dataBase := redis.NewDatabase(logger, databaseSettings, redis.NotificationHistoryConfig{}, redis.Cli)
+	dataBase := redis.NewDatabase(logger, databaseSettings, redis.NotificationHistoryConfig{}, redis.NotificationConfig{}, redis.Cli)
 	return config.Cleanup, logger, dataBase
 }
 

cmd/config.go

@@ -79,6 +79,30 @@ func (notificationHistoryConfig *NotificationHistoryConfig) GetSettings() redis.
 	}
 }
 
+// NotificationConfig is a config that stores the necessary configuration of the notifier
+type NotificationConfig struct {
+	// Need to determine if notification is delayed - the difference between creation time and sending time
+	// is greater than DelayedTime
+	DelayedTime string `yaml:"delayed_time"`
+	// TransactionTimeout defines the timeout between fetch notifications transactions
+	TransactionTimeout string `yaml:"transaction_timeout"`
+	// TransactionMaxRetries defines the maximum number of attempts to make a transaction
+	TransactionMaxRetries int `yaml:"transaction_max_retries"`
+	// TransactionHeuristicLimit maximum allowable limit, after this limit all notifications
+	// without limit will be taken
+	TransactionHeuristicLimit int64 `yaml:"transaction_heuristic_limit"`
+}
+
+// GetSettings returns notification storage configuration
+func (notificationConfig *NotificationConfig) GetSettings() redis.NotificationConfig {
+	return redis.NotificationConfig{
+		DelayedTime:               to.Duration(notificationConfig.DelayedTime),
+		TransactionTimeout:        to.Duration(notificationConfig.TransactionTimeout),
+		TransactionMaxRetries:     notificationConfig.TransactionMaxRetries,
+		TransactionHeuristicLimit: notificationConfig.TransactionHeuristicLimit,
+	}
+}
+
 // GraphiteConfig is graphite metrics config structure that initialises at the start of moira
 type GraphiteConfig struct {
 	// If true, graphite sender will be enabled.

cmd/filter/main.go

@@ -87,7 +87,7 @@ func main() {
 	}
 
 	filterMetrics := metrics.ConfigureFilterMetrics(telemetry.Metrics)
-	database := redis.NewDatabase(logger, config.Redis.GetSettings(), redis.NotificationHistoryConfig{}, redis.Filter)
+	database := redis.NewDatabase(logger, config.Redis.GetSettings(), redis.NotificationHistoryConfig{}, redis.NotificationConfig{}, redis.Filter)
 
 	retentionConfigFile, err := os.Open(config.Filter.RetentionConfig)
 	if err != nil {

cmd/notifier/config.go

@@ -21,6 +21,7 @@ type config struct {
 	Prometheus          cmd.PrometheusConfig          `yaml:"prometheus"`
 	ImageStores         cmd.ImageStoreConfig          `yaml:"image_store"`
 	NotificationHistory cmd.NotificationHistoryConfig `yaml:"notification_history"`
+	Notification        cmd.NotificationConfig        `yaml:"notification"`
 }
 
 type entityLogConfig struct {
@@ -93,6 +94,12 @@ func getDefault() config {
 			NotificationHistoryTTL:        "48h",
 			NotificationHistoryQueryLimit: int(notifier.NotificationsLimitUnlimited),
 		},
+		Notification: cmd.NotificationConfig{
+			DelayedTime:               "1m",
+			TransactionTimeout:        "200ms",
+			TransactionMaxRetries:     10,
+			TransactionHeuristicLimit: 10000,
+		},
 		Notifier: notifierConfig{
 			SenderTimeout:    "10s",
 			ResendingTimeout: "1:00",

cmd/notifier/main.go

@@ -81,7 +81,8 @@ func main() {
 
 	databaseSettings := config.Redis.GetSettings()
 	notificationHistorySettings := config.NotificationHistory.GetSettings()
-	database := redis.NewDatabase(logger, databaseSettings, notificationHistorySettings, redis.Notifier)
+	notificationSettings := config.Notification.GetSettings()
+	database := redis.NewDatabase(logger, databaseSettings, notificationHistorySettings, notificationSettings, redis.Notifier)
 
 	remoteConfig := config.Remote.GetRemoteSourceSettings()
 	prometheusConfig := config.Prometheus.GetPrometheusSourceSettings()

database/redis/config.go

@@ -21,3 +21,17 @@ type NotificationHistoryConfig struct {
 	NotificationHistoryTTL        time.Duration
 	NotificationHistoryQueryLimit int
 }
+
+// Notifier configuration in redis
+type NotificationConfig struct {
+	// Need to determine if notification is delayed - the difference between creation time and sending time
+	// is greater than DelayedTime
+	DelayedTime time.Duration
+	// TransactionTimeout defines the timeout between fetch notifications transactions
+	TransactionTimeout time.Duration
+	// TransactionMaxRetries defines the maximum number of attempts to make a transaction
+	TransactionMaxRetries int
+	// TransactionHeuristicLimit maximum allowable limit, after this limit all notifications
+	// without limit will be taken
+	TransactionHeuristicLimit int64
+}

database/redis/contact_notification_history_test.go

@@ -46,6 +46,7 @@ var inputScheduledNotification = moira.ScheduledNotification{
 	Throttled: false,
 	SendFail:  1,
 	Timestamp: time.Now().Unix(),
+	CreatedAt: time.Now().Unix(),
 }
 
 var eventsShouldBeInDb = []*moira.NotificationEventHistoryItem{

database/redis/database.go

@@ -46,9 +46,11 @@ type DbConnector struct {
 	source               DBSource
 	clock                moira.Clock
 	notificationHistory  NotificationHistoryConfig
+	// Notifier configuration in redis
+	notification NotificationConfig
 }
 
-func NewDatabase(logger moira.Logger, config DatabaseConfig, nh NotificationHistoryConfig, source DBSource) *DbConnector {
+func NewDatabase(logger moira.Logger, config DatabaseConfig, nh NotificationHistoryConfig, n NotificationConfig, source DBSource) *DbConnector {
 	client := redis.NewUniversalClient(&redis.UniversalOptions{
 		MasterName:       config.MasterName,
 		Addrs:            config.Addrs,
@@ -78,7 +80,9 @@ func NewDatabase(logger moira.Logger, config DatabaseConfig, nh NotificationHist
 		source:               source,
 		clock:                clock.NewSystemClock(),
 		notificationHistory:  nh,
+		notification:         n,
 	}
+
 	return &connector
 }
 
@@ -90,7 +94,14 @@ func NewTestDatabase(logger moira.Logger) *DbConnector {
 		NotificationHistoryConfig{
 			NotificationHistoryTTL:        time.Hour * 48,
 			NotificationHistoryQueryLimit: 1000,
-		}, testSource)
+		},
+		NotificationConfig{
+			DelayedTime:               time.Minute,
+			TransactionTimeout:        200 * time.Millisecond,
+			TransactionMaxRetries:     10,
+			TransactionHeuristicLimit: 10000,
+		},
+		testSource)
 }
 
 // NewTestDatabaseWithIncorrectConfig use it only for tests
@@ -101,6 +112,12 @@ func NewTestDatabaseWithIncorrectConfig(logger moira.Logger) *DbConnector {
 			NotificationHistoryTTL:        time.Hour * 48,
 			NotificationHistoryQueryLimit: 1000,
 		},
+		NotificationConfig{
+			DelayedTime:               time.Minute,
+			TransactionTimeout:        200 * time.Millisecond,
+			TransactionMaxRetries:     10,
+			TransactionHeuristicLimit: 10000,
+		},
 		testSource)
 }
 

database/redis/last_check.go

@@ -198,6 +198,28 @@ func (connector *DbConnector) checkDataScoreChanged(triggerID string, checkData
 	return oldScore != float64(checkData.Score)
 }
 
+// getTriggersLastCheck returns an array of trigger checks by the passed ids, if the trigger does not exist, it is nil
+func (connector *DbConnector) getTriggersLastCheck(triggerIDs []string) ([]*moira.CheckData, error) {
+	ctx := connector.context
+	pipe := (*connector.client).TxPipeline()
+
+	results := make([]*redis.StringCmd, len(triggerIDs))
+	for i, id := range triggerIDs {
+		var result *redis.StringCmd
+		if id != "" {
+			result = pipe.Get(ctx, metricLastCheckKey(id))
+		}
+		results[i] = result
+	}
+
+	_, err := pipe.Exec(ctx)
+	if err != nil && err != redis.Nil {
+		return nil, err
+	}
+
+	return reply.Checks(results)
+}
+
 var badStateTriggersKey = "moira-bad-state-triggers"
 var triggersChecksKey = "moira-triggers-checks"
 

database/redis/last_check_test.go

@@ -469,6 +469,107 @@ func TestLastCheckErrorConnection(t *testing.T) {
 	})
 }
 
+func TestGetTriggersLastCheck(t *testing.T) {
+	logger, _ := logging.GetLogger("dataBase")
+	dataBase := NewTestDatabase(logger)
+	dataBase.Flush()
+	defer dataBase.Flush()
+
+	_ = dataBase.SetTriggerLastCheck("test1", &moira.CheckData{
+		Timestamp: 1,
+	}, moira.TriggerSourceNotSet)
+
+	_ = dataBase.SetTriggerLastCheck("test2", &moira.CheckData{
+		Timestamp: 2,
+	}, moira.TriggerSourceNotSet)
+
+	_ = dataBase.SetTriggerLastCheck("test3", &moira.CheckData{
+		Timestamp: 3,
+	}, moira.TriggerSourceNotSet)
+
+	Convey("getTriggersLastCheck manipulations", t, func() {
+		Convey("Test with nil id array", func() {
+			actual, err := dataBase.getTriggersLastCheck(nil)
+			So(err, ShouldBeNil)
+			So(actual, ShouldResemble, []*moira.CheckData{})
+		})
+
+		Convey("Test with correct id array", func() {
+			actual, err := dataBase.getTriggersLastCheck([]string{"test1", "test2", "test3"})
+			So(err, ShouldBeNil)
+			So(actual, ShouldResemble, []*moira.CheckData{
+				{
+					Timestamp:               1,
+					MetricsToTargetRelation: map[string]string{},
+				},
+				{
+					Timestamp:               2,
+					MetricsToTargetRelation: map[string]string{},
+				},
+				{
+					Timestamp:               3,
+					MetricsToTargetRelation: map[string]string{},
+				},
+			})
+		})
+
+		Convey("Test with deleted trigger", func() {
+			dataBase.RemoveTriggerLastCheck("test2") //nolint
+			defer func() {
+				_ = dataBase.SetTriggerLastCheck("test2", &moira.CheckData{
+					Timestamp: 2,
+				}, moira.TriggerSourceNotSet)
+			}()
+
+			actual, err := dataBase.getTriggersLastCheck([]string{"test1", "test2", "test3"})
+			So(err, ShouldBeNil)
+			So(actual, ShouldResemble, []*moira.CheckData{
+				{
+					Timestamp:               1,
+					MetricsToTargetRelation: map[string]string{},
+				},
+				nil,
+				{
+					Timestamp:               3,
+					MetricsToTargetRelation: map[string]string{},
+				},
+			})
+		})
+
+		Convey("Test with a nonexistent trigger id", func() {
+			actual, err := dataBase.getTriggersLastCheck([]string{"test1", "test2", "test4"})
+			So(err, ShouldBeNil)
+			So(actual, ShouldResemble, []*moira.CheckData{
+				{
+					Timestamp:               1,
+					MetricsToTargetRelation: map[string]string{},
+				},
+				{
+					Timestamp:               2,
+					MetricsToTargetRelation: map[string]string{},
+				},
+				nil,
+			})
+		})
+
+		Convey("Test with an empty trigger id", func() {
+			actual, err := dataBase.getTriggersLastCheck([]string{"", "test2", "test3"})
+			So(err, ShouldBeNil)
+			So(actual, ShouldResemble, []*moira.CheckData{
+				nil,
+				{
+					Timestamp:               2,
+					MetricsToTargetRelation: map[string]string{},
+				},
+				{
+					Timestamp:               3,
+					MetricsToTargetRelation: map[string]string{},
+				},
+			})
+		})
+	})
+}
+
 func TestMaintenanceUserSave(t *testing.T) {
 	logger, _ := logging.GetLogger("dataBase")
 	dataBase := NewTestDatabase(logger)