feat(notifier): add ability to scale webhook and script

cmd/notifier/config.go

@@ -55,6 +55,8 @@ type notifierConfig struct {
 	MaxFailAttemptToSendAvailable int `yaml:"max_fail_attempt_to_send_available"`
 	// Specify log level by entities
 	SetLogLevel setLogLevelConfig `yaml:"set_log_level"`
+	// Sets the maximum number of sends per sender
+	MaxParallelSendsPerSender int `yaml:"max_parallel_sends_per_sender"`
 }
 
 type selfStateConfig struct {
@@ -114,6 +116,7 @@ func getDefault() config {
 			Timezone:                      "UTC",
 			ReadBatchSize:                 int(notifier.NotificationsLimitUnlimited),
 			MaxFailAttemptToSendAvailable: 3,
+			MaxParallelSendsPerSender:     16,
 		},
 		Telemetry: cmd.TelemetryConfig{
 			Listen: ":8093",
@@ -208,6 +211,7 @@ func (config *notifierConfig) getSettings(logger moira.Logger) notifier.Config {
 		MaxFailAttemptToSendAvailable: config.MaxFailAttemptToSendAvailable,
 		LogContactsToLevel:            contacts,
 		LogSubscriptionsToLevel:       subscriptions,
+		MaxParallelSendsPerSender:     config.MaxParallelSendsPerSender,
 	}
 }
 

cmd/notifier/main.go

@@ -104,13 +104,15 @@ func main() {
 	notifierConfig := config.Notifier.getSettings(logger)
 
 	notifierMetrics := metrics.ConfigureNotifierMetrics(telemetry.Metrics, serviceName)
+	scheduler := notifier.NewScheduler(database, logger, notifierMetrics)
 	sender := notifier.NewNotifier(
 		database,
 		logger,
 		notifierConfig,
 		notifierMetrics,
 		metricSourceProvider,
 		imageStoreMap,
+		scheduler,
 	)
 
 	// Register moira senders

database/redis/trigger.go

@@ -208,7 +208,7 @@ func (connector *DbConnector) GetTriggerIDsStartWith(prefix string) ([]string, e
 	return matchedTriggers, nil
 }
 
-func (connector *DbConnector) updateTrigger(triggerID string, newTrigger *moira.Trigger, oldTrigger *moira.Trigger) error {
+func (connector *DbConnector) updateTrigger(triggerID string, newTrigger *moira.Trigger, oldTrigger *moira.Trigger) error { // nolint:gocyclo
 	bytes, err := reply.GetTriggerBytes(triggerID, newTrigger)
 	if err != nil {
 		return err

integration_tests/notifier/notifier_test.go

@@ -27,11 +27,12 @@ var location, _ = time.LoadLocation("UTC")
 var dateTimeFormat = "15:04 02.01.2006"
 
 var notifierConfig = notifier.Config{
-	SendingTimeout:   time.Millisecond * 10,
-	ResendingTimeout: time.Hour * 24,
-	Location:         location,
-	DateTimeFormat:   dateTimeFormat,
-	ReadBatchSize:    notifier.NotificationsLimitUnlimited,
+	SendingTimeout:            time.Millisecond * 10,
+	ResendingTimeout:          time.Hour * 24,
+	Location:                  location,
+	DateTimeFormat:            dateTimeFormat,
+	ReadBatchSize:             notifier.NotificationsLimitUnlimited,
+	MaxParallelSendsPerSender: 16,
 }
 
 var shutdown = make(chan struct{})
@@ -87,6 +88,7 @@ func TestNotifier(t *testing.T) {
 	database.PushNotificationEvent(&event, true) //nolint
 
 	metricsSourceProvider := metricSource.CreateMetricSourceProvider(local.Create(database), nil, nil)
+	scheduler := notifier.NewScheduler(database, logger, notifierMetrics)
 
 	notifierInstance := notifier.NewNotifier(
 		database,
@@ -95,6 +97,7 @@ func TestNotifier(t *testing.T) {
 		notifierMetrics,
 		metricsSourceProvider,
 		map[string]moira.ImageStore{},
+		scheduler,
 	)
 
 	sender := mock_moira_alert.NewMockSender(mockCtrl)
@@ -113,7 +116,7 @@ func TestNotifier(t *testing.T) {
 		Database:  database,
 		Logger:    logger,
 		Metrics:   notifierMetrics,
-		Scheduler: notifier.NewScheduler(database, logger, notifierMetrics),
+		Scheduler: scheduler,
 	}
 
 	fetchNotificationsWorker := notifications.FetchNotificationsWorker{

local/notifier.yml

@@ -39,6 +39,7 @@ notifier:
   front_uri: http://localhost
   timezone: UTC
   date_time_format: "15:04 02.01.2006"
+  max_parallel_sends_per_sender: 16
 notification_history:
   ttl: 48h
   query_limit: 10000

notifier/config.go

@@ -23,4 +23,5 @@ type Config struct {
 	MaxFailAttemptToSendAvailable int
 	LogContactsToLevel            map[string]string
 	LogSubscriptionsToLevel       map[string]string
+	MaxParallelSendsPerSender     int
 }

notifier/notifier.go

@@ -73,38 +73,59 @@ type Notifier interface {
 
 // StandardNotifier represent notification functionality
 type StandardNotifier struct {
-	waitGroup            sync.WaitGroup
-	senders              map[string]chan NotificationPackage
-	logger               moira.Logger
-	database             moira.Database
-	scheduler            Scheduler
-	config               Config
-	metrics              *metrics.NotifierMetrics
-	metricSourceProvider *metricSource.SourceProvider
-	imageStores          map[string]moira.ImageStore
+	waitGroup              sync.WaitGroup
+	senders                map[string]moira.Sender
+	sendersNotificationsCh map[string]chan NotificationPackage
+	logger                 moira.Logger
+	database               moira.Database
+	scheduler              Scheduler
+	config                 Config
+	metrics                *metrics.NotifierMetrics
+	metricSourceProvider   *metricSource.SourceProvider
+	imageStores            map[string]moira.ImageStore
+	sendersNameToType      map[string]string
+	sendersOnce            map[string]*sync.Once
 }
 
 // NewNotifier is initializer for StandardNotifier
-func NewNotifier(database moira.Database, logger moira.Logger, config Config, metrics *metrics.NotifierMetrics, metricSourceProvider *metricSource.SourceProvider, imageStoreMap map[string]moira.ImageStore) *StandardNotifier {
+func NewNotifier(
+	database moira.Database,
+	logger moira.Logger,
+	config Config,
+	metrics *metrics.NotifierMetrics,
+	metricSourceProvider *metricSource.SourceProvider,
+	imageStoreMap map[string]moira.ImageStore,
+	scheduler Scheduler,
+) *StandardNotifier {
 	return &StandardNotifier{
-		senders:              make(map[string]chan NotificationPackage),
-		logger:               logger,
-		database:             database,
-		scheduler:            NewScheduler(database, logger, metrics),
-		config:               config,
-		metrics:              metrics,
-		metricSourceProvider: metricSourceProvider,
-		imageStores:          imageStoreMap,
+		senders:                make(map[string]moira.Sender),
+		sendersNotificationsCh: make(map[string]chan NotificationPackage),
+		logger:                 logger,
+		database:               database,
+		scheduler:              scheduler,
+		config:                 config,
+		metrics:                metrics,
+		metricSourceProvider:   metricSourceProvider,
+		imageStores:            imageStoreMap,
+		sendersNameToType:      make(map[string]string),
+		sendersOnce:            make(map[string]*sync.Once),
 	}
 }
 
 // Send is realization of StandardNotifier Send functionality
 func (notifier *StandardNotifier) Send(pkg *NotificationPackage, waitGroup *sync.WaitGroup) {
-	ch, found := notifier.senders[pkg.Contact.Type]
-	if !found {
+	senderType, ok := notifier.sendersNameToType[pkg.Contact.Type]
+	if !ok {
 		notifier.resend(pkg, fmt.Sprintf("Unknown contact type '%s' [%s]", pkg.Contact.Type, pkg))
 		return
 	}
+
+	ch, found := notifier.sendersNotificationsCh[senderType]
+	if !found {
+		notifier.resend(pkg, fmt.Sprintf("failed to get notification channel from '%s' [%s]", senderType, pkg))
+		return
+	}
+
 	waitGroup.Add(1)
 	go func(pkg *NotificationPackage) {
 		defer waitGroup.Done()
@@ -142,6 +163,7 @@ func (notifier *StandardNotifier) resend(pkg *NotificationPackage, reason string
 		notifier.metrics.MarkSendersDroppedNotifications(pkg.Contact.Type)
 		return
 	}
+
 	notifier.metrics.SendingFailed.Mark(1)
 	if metric, found := notifier.metrics.SendersFailedMetrics.GetRegisteredMeter(pkg.Contact.Type); found {
 		metric.Mark(1)
@@ -212,6 +234,7 @@ func (notifier *StandardNotifier) runSender(sender moira.Sender, ch chan Notific
 			notifier.metrics.MarkSendersOkMetrics(pkg.Contact.Type)
 			continue
 		}
+
 		switch e := err.(type) {
 		case moira.SenderBrokenContactError:
 			log.Warning().

notifier/notifier_test.go

@@ -19,17 +19,26 @@ import (
 )
 
 var (
-	plots    [][]byte
-	shutdown = make(chan struct{})
+	plots          [][]byte
+	shutdown       = make(chan struct{})
+	location, _    = time.LoadLocation("UTC")
+	dateTimeFormat = "15:04 02.01.2006"
+	defaultConfig  = Config{
+		SendingTimeout:            time.Millisecond * 10,
+		ResendingTimeout:          time.Hour * 24,
+		Location:                  location,
+		DateTimeFormat:            dateTimeFormat,
+		MaxParallelSendsPerSender: 16,
+	}
 )
 
 var (
-	mockCtrl  *gomock.Controller
-	sender    *mock_moira_alert.MockSender
-	notif     *StandardNotifier
-	scheduler *mock_scheduler.MockScheduler
-	dataBase  *mock_moira_alert.MockDatabase
-	logger    moira.Logger
+	mockCtrl         *gomock.Controller
+	sender           *mock_moira_alert.MockSender
+	standardNotifier *StandardNotifier
+	scheduler        *mock_scheduler.MockScheduler
+	dataBase         *mock_moira_alert.MockDatabase
+	logger           moira.Logger
 )
 
 func TestGetMetricNames(t *testing.T) {
@@ -39,6 +48,7 @@ func TestGetMetricNames(t *testing.T) {
 			actual := notificationsPackage.GetMetricNames()
 			So(actual, ShouldResemble, expected)
 		})
+
 		Convey("Test package with no trigger events", func() {
 			pkg := NotificationPackage{}
 			for _, event := range notificationsPackage.Events {
@@ -52,6 +62,7 @@ func TestGetMetricNames(t *testing.T) {
 			So(actual, ShouldResemble, expected)
 		})
 	})
+
 	Convey("Test empty notification package", t, func() {
 		emptyNotificationPackage := NotificationPackage{}
 		actual := emptyNotificationPackage.GetMetricNames()
@@ -66,6 +77,7 @@ func TestGetWindow(t *testing.T) {
 		So(from, ShouldEqual, 11)
 		So(to, ShouldEqual, 179)
 	})
+
 	Convey("Test empty notification package", t, func() {
 		emptyNotificationPackage := NotificationPackage{}
 		_, _, err := emptyNotificationPackage.GetWindow()
@@ -74,7 +86,7 @@ func TestGetWindow(t *testing.T) {
 }
 
 func TestUnknownContactType(t *testing.T) {
-	configureNotifier(t)
+	configureNotifier(t, defaultConfig)
 	defer afterTest()
 
 	var eventsData moira.NotificationEvents = []moira.NotificationEvent{event}
@@ -90,12 +102,12 @@ func TestUnknownContactType(t *testing.T) {
 	dataBase.EXPECT().AddNotification(&notification).Return(nil)
 
 	var wg sync.WaitGroup
-	notif.Send(&pkg, &wg)
+	standardNotifier.Send(&pkg, &wg)
 	wg.Wait()
 }
 
 func TestFailSendEvent(t *testing.T) {
-	configureNotifier(t)
+	configureNotifier(t, defaultConfig)
 	defer afterTest()
 
 	var eventsData moira.NotificationEvents = []moira.NotificationEvent{event}
@@ -106,13 +118,14 @@ func TestFailSendEvent(t *testing.T) {
 			Type: "test",
 		},
 	}
+
 	notification := moira.ScheduledNotification{}
 	sender.EXPECT().SendEvents(eventsData, pkg.Contact, pkg.Trigger, plots, pkg.Throttled).Return(fmt.Errorf("Cant't send"))
 	scheduler.EXPECT().ScheduleNotification(gomock.Any(), event, pkg.Trigger, pkg.Contact, pkg.Plotting, pkg.Throttled, pkg.FailCount+1, gomock.Any()).Return(&notification)
 	dataBase.EXPECT().AddNotification(&notification).Return(nil)
 
 	var wg sync.WaitGroup
-	notif.Send(&pkg, &wg)
+	standardNotifier.Send(&pkg, &wg)
 	wg.Wait()
 	time.Sleep(time.Second * 2)
 }
@@ -122,7 +135,7 @@ func TestNoResendForSendToBrokenContact(t *testing.T) {
 		t.Skip("skipping test in short mode.")
 	}
 
-	configureNotifier(t)
+	configureNotifier(t, defaultConfig)
 	defer afterTest()
 
 	var eventsData moira.NotificationEvents = []moira.NotificationEvent{event}
@@ -137,13 +150,13 @@ func TestNoResendForSendToBrokenContact(t *testing.T) {
 		Return(moira.NewSenderBrokenContactError(fmt.Errorf("some sender reason")))
 
 	var wg sync.WaitGroup
-	notif.Send(&pkg, &wg)
+	standardNotifier.Send(&pkg, &wg)
 	wg.Wait()
 	time.Sleep(time.Second * 2)
 }
 
 func TestTimeout(t *testing.T) {
-	configureNotifier(t)
+	configureNotifier(t, defaultConfig)
 	var wg sync.WaitGroup
 	defer afterTest()
 
@@ -160,10 +173,10 @@ func TestTimeout(t *testing.T) {
 	sender.EXPECT().SendEvents(eventsData, pkg.Contact, pkg.Trigger, plots, pkg.Throttled).Return(nil).Do(func(arg0, arg1, arg2, arg3, arg4 interface{}) {
 		fmt.Print("Trying to send for 10 second")
 		time.Sleep(time.Second * 10)
-	}).Times(maxParallelSendsPerSender)
+	}).Times(standardNotifier.GetMaxParallelSendsPerSender())
 
-	for i := 0; i < maxParallelSendsPerSender; i++ {
-		notif.Send(&pkg, &wg)
+	for i := 0; i < standardNotifier.GetMaxParallelSendsPerSender(); i++ {
+		standardNotifier.Send(&pkg, &wg)
 		wg.Wait()
 	}
 
@@ -180,7 +193,7 @@ func TestTimeout(t *testing.T) {
 	scheduler.EXPECT().ScheduleNotification(gomock.Any(), event, pkg2.Trigger, pkg2.Contact, pkg.Plotting, pkg2.Throttled, pkg2.FailCount+1, gomock.Any()).Return(&notification)
 	dataBase.EXPECT().AddNotification(&notification).Return(nil).Do(func(f ...interface{}) { close(shutdown) })
 
-	notif.Send(&pkg2, &wg)
+	standardNotifier.Send(&pkg2, &wg)
 	wg.Wait()
 	waitTestEnd()
 }
@@ -195,16 +208,8 @@ func waitTestEnd() {
 	}
 }
 
-func configureNotifier(t *testing.T) {
+func configureNotifier(t *testing.T, config Config) {
 	notifierMetrics := metrics.ConfigureNotifierMetrics(metrics.NewDummyRegistry(), "notifier")
-	var location, _ = time.LoadLocation("UTC")
-	dateTimeFormat := "15:04 02.01.2006"
-	config := Config{
-		SendingTimeout:   time.Millisecond * 10,
-		ResendingTimeout: time.Hour * 24,
-		Location:         location,
-		DateTimeFormat:   dateTimeFormat,
-	}
 
 	mockCtrl = gomock.NewController(t)
 	dataBase = mock_moira_alert.NewMockDatabase(mockCtrl)
@@ -213,24 +218,25 @@ func configureNotifier(t *testing.T) {
 	sender = mock_moira_alert.NewMockSender(mockCtrl)
 	metricsSourceProvider := metricSource.CreateMetricSourceProvider(local.Create(dataBase), nil, nil)
 
-	notif = NewNotifier(dataBase, logger, config, notifierMetrics, metricsSourceProvider, map[string]moira.ImageStore{})
-	notif.scheduler = scheduler
+	standardNotifier = NewNotifier(dataBase, logger, config, notifierMetrics, metricsSourceProvider, map[string]moira.ImageStore{}, scheduler)
 	senderSettings := map[string]interface{}{
 		"type": "test",
 	}
 
-	sender.EXPECT().Init(senderSettings, logger, location, "15:04 02.01.2006").Return(nil)
+	sender.EXPECT().Init(senderSettings, logger, location, dateTimeFormat).Return(nil)
 
-	notif.RegisterSender(senderSettings, sender) //nolint
+	err := standardNotifier.RegisterSender(senderSettings, sender)
 
 	Convey("Should return one sender", t, func() {
-		So(notif.GetSenders(), ShouldResemble, map[string]bool{"test": true})
+		So(err, ShouldBeNil)
+		So(standardNotifier.GetSenders(), ShouldResemble, map[string]bool{"test": true})
+		So(standardNotifier.sendersNameToType["test"], ShouldEqual, senderSettings["type"])
 	})
 }
 
 func afterTest() {
 	mockCtrl.Finish()
-	notif.StopSenders()
+	standardNotifier.StopSenders()
 }
 
 var subID = "SubscriptionID-000000000000001"