Reduce notification delay

pkg/csplugin/broker_test.go

@@ -297,8 +297,6 @@ func (s *PluginSuite) TestBrokerRunGroupThreshold() {
 	// because of group threshold, we shouldn't have data yet
 	assert.NoFileExists(t, s.outFile)
 	pb.PluginChannel <- models.ProfileAlert{ProfileID: uint(0), Alert: &models.Alert{}}
-	pb.PluginChannel <- models.ProfileAlert{ProfileID: uint(0), Alert: &models.Alert{}}
-	pb.PluginChannel <- models.ProfileAlert{ProfileID: uint(0), Alert: &models.Alert{}}
 
 	time.Sleep(time.Second)
 
@@ -310,16 +308,12 @@ func (s *PluginSuite) TestBrokerRunGroupThreshold() {
 
 	var alerts []models.Alert
 
-	// two notifications, one with 4 alerts, one with 2 alerts
+	// one notification with 4 alerts
 
 	err = decoder.Decode(&alerts)
 	require.NoError(t, err)
 	assert.Len(t, alerts, 4)
 
-	err = decoder.Decode(&alerts)
-	require.NoError(t, err)
-	assert.Len(t, alerts, 2)
-
 	err = decoder.Decode(&alerts)
 	assert.Equal(t, err, io.EOF)
 }

pkg/csplugin/watcher.go

@@ -160,9 +160,22 @@ func (pw *PluginWatcher) watchPluginAlertCounts() {
 		select {
 		case pluginName := <-pw.Inserts:
 			// we only "count" pending alerts, and watchPluginTicker is actually going to send it
-			if _, ok := pw.PluginConfigByName[pluginName]; ok {
+			if cfg, ok := pw.PluginConfigByName[pluginName]; ok {
 				// atomic increment, prevent race between Get/Set
 				pw.AlertCountByPluginName.Inc(pluginName)
+
+				// threshold reached: flush now instead of at the next tick. In a goroutine,
+				// the broker may be blocked sending us the next insert.
+				if pc, _ := pw.AlertCountByPluginName.Get(pluginName); cfg.GroupThreshold > 0 && pc >= cfg.GroupThreshold {
+					pw.AlertCountByPluginName.Set(pluginName, 0)
+
+					go func() {
+						select {
+						case pw.PluginEvents <- pluginName:
+						case <-pw.tomb.Dead():
+						}
+					}()
+				}
 			}
 		case <-pw.tomb.Dying():
 			return

pkg/csplugin/watcher_test.go

@@ -127,3 +127,39 @@ func TestPluginAlertCountWatcher(t *testing.T) {
 	require.NoError(t, err)
 	resetTestTomb(t, &testTomb, &pw)
 }
+
+func TestPluginAlertCountWatcherFlushesImmediately(t *testing.T) {
+	cstest.SkipOnWindowsBecause(t, "timing is not reliable")
+
+	// only an immediate flush can get the event out before the first tick
+	oldTicker := DefaultEmptyTicker
+	DefaultEmptyTicker = time.Hour
+
+	t.Cleanup(func() { DefaultEmptyTicker = oldTicker })
+
+	tests := []struct {
+		name   string
+		config PluginConfig
+	}{
+		{"threshold only", PluginConfig{GroupThreshold: 1}},
+		{"threshold and long interval", PluginConfig{GroupThreshold: 1, GroupWait: time.Hour}},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			pw := PluginWatcher{}
+			testTomb := tomb.Tomb{}
+			pw.Init(map[string]PluginConfig{"testPlugin": tc.config}, make(map[string][]*models.Alert))
+			pw.Start(&testTomb)
+
+			insertNAlertsToPlugin(&pw, 1, "testPlugin")
+
+			ct, cancel := context.WithTimeout(t.Context(), 200*time.Millisecond)
+			defer cancel()
+
+			err := listenChannelWithTimeout(ct, pw.PluginEvents)
+			require.NoError(t, err)
+			resetTestTomb(t, &testTomb, &pw)
+		})
+	}
+}