feat: add notification body on send errors & add test to ensure notification.send events are retried

notification/notification_test.go

@@ -23,6 +23,134 @@
 	_ "github.com/flanksource/incident-commander/upstream"
 )
 
+var _ = ginkgo.Describe("notification error handling on send", ginkgo.Ordered, func() {
+	var goodNotif models.Notification
+	var badNotif models.Notification
+	var deployment1 models.ConfigItem
+	var pod1 models.ConfigItem
+
+	ginkgo.BeforeAll(func() {
+		{
+			customReceiver := []api.NotificationConfig{
+				{
+					URL: fmt.Sprintf("generic+%s", webhookEndpoint),
+					Properties: map[string]string{
+						"disabletls": "yes",
+						"template":   "json",
+					},
+				},
+			}
+			customReceiverJson, err := json.Marshal(customReceiver)
+			Expect(err).To(BeNil())
+
+			goodNotif = models.Notification{
+				ID:             uuid.New(),
+				Name:           "test-notif-1",
+				Events:         pq.StringArray([]string{"config.updated"}),
+				Filter:         ".config.type == 'Kubernetes::Deployment'",
+				Source:         models.SourceCRD,
+				Title:          "Dummy",
+				Template:       "dummy",
+				CustomServices: types.JSON(customReceiverJson),
+			}
+
+			err = DefaultContext.DB().Create(&goodNotif).Error
+			Expect(err).To(BeNil())
+		}
+
+		{
+			badReceiver := []api.NotificationConfig{
+				{
+					URL: "generic+bad",
+					Properties: map[string]string{
+						"disabletls": "yes",
+						"template":   "json",
+					},
+				},
+			}
+			customReceiverJson, err := json.Marshal(badReceiver)
+			Expect(err).To(BeNil())
+
+			badNotif = models.Notification{
+				ID:             uuid.New(),
+				Name:           "test-notif-2",
+				Events:         pq.StringArray([]string{"config.updated"}),
+				Filter:         ".config.type == 'Kubernetes::Pod'",
+				Source:         models.SourceCRD,
+				Title:          "Dummy",
+				Template:       "dummy",
+				CustomServices: types.JSON(customReceiverJson),
+			}
+
+			err = DefaultContext.DB().Create(&badNotif).Error
+			Expect(err).To(BeNil())
+		}
+
+		{
+			deployment1 = models.ConfigItem{
+				ID:          uuid.New(),
+				Name:        lo.ToPtr("deployment-1"),
+				ConfigClass: models.ConfigClassDeployment,
+				Config:      lo.ToPtr(`{"replicas": 1}`),
+				Type:        lo.ToPtr("Kubernetes::Deployment"),
+			}
+
+			err := DefaultContext.DB().Create(&deployment1).Error
+			Expect(err).To(BeNil())
+		}
+
+		{
+			pod1 = models.ConfigItem{
+				ID:          uuid.New(),
+				Name:        lo.ToPtr("deployment-2"),
+				ConfigClass: models.ConfigClassDeployment,
+				Config:      lo.ToPtr(`{"replicas": 2}`),
+				Type:        lo.ToPtr("Kubernetes::Pod"),
+			}
+
+			err := DefaultContext.DB().Create(&pod1).Error
+			Expect(err).To(BeNil())
+		}
+	})
+
+	ginkgo.It("should have consumed all events", func() {
+		testEvents := []models.Event{
+			{
+				Name:       "config.updated",
+				Properties: types.JSONStringMap{"id": deployment1.ID.String()},
+			}, {
+				Name:       "config.updated",
+				Properties: types.JSONStringMap{"id": pod1.ID.String()},
+			},
+		}
+		err := DefaultContext.DB().Create(&testEvents).Error
+		Expect(err).To(BeNil())
+
+		events.ConsumeAll(DefaultContext)
+		Eventually(func() int64 {
+			var c int64
+			DefaultContext.DB().Model(&models.Event{}).Where("name = 'config.updated'").Count(&c)
+			return c
+		}, "10s", "200ms").Should(Equal(int64(0)))
+	})
+
+	ginkgo.It("one notification.send event with max attempt should be in the event_queue", func() {
+		Eventually(func() int {
+			var event models.Event
+			err := DefaultContext.DB().Where("name = 'notification.send'").First(&event).Error
+			Expect(err).To(BeNil())
+			return event.Attempts
+		}, "10s", "200ms").Should(Equal(4))
+	})
+
+	ginkgo.It("only one notification must have been sent", func() {
+		var sentHistoryCount int64
+		err := DefaultContext.DB().Model(&models.NotificationSendHistory{}).Where("notification_id = ?", goodNotif.ID).Count(&sentHistoryCount).Error
+		Expect(err).To(BeNil())
+		Expect(sentHistoryCount).To(Equal(int64(1)))
+	})
+})
+
 var _ = ginkgo.Describe("Notifications", ginkgo.Ordered, func() {
 	var _ = ginkgo.Describe("Notification on incident creation", ginkgo.Ordered, func() {
 		var (
@@ -95,7 +223,7 @@
 			}
 
 			err := DefaultContext.DB().Create(&notif).Error
 			Expect(err).To(BeNil())
 		})
 
 		ginkgo.It("should create an incident", func() {

notification/send.go

@@ -162,7 +162,12 @@ func SendNotification(ctx *Context, connectionName, shoutrrrURL string, celEnv m
 		if err := templater.Walk(&data); err != nil {
 			return "", fmt.Errorf("error templating notification: %w", err)
 		}
-		return "slack", SlackSend(ctx, connection.Password, connection.Username, data)
+
+		if err := SlackSend(ctx, connection.Password, connection.Username, data); err != nil {
+			return "", fmt.Errorf("failed to send message (%s) to slack: %w", data.Message, err)
+		}
+
+		return "slack", nil
 	}
 
 	service, err := shoutrrrSend(ctx, celEnv, shoutrrrURL, data)

notification/shoutrrr.go

@@ -117,7 +117,7 @@ func shoutrrrSend(ctx *Context, celEnv map[string]any, shoutrrrURL string, data
 	sendErrors := sender.Send(data.Message, params)
 	for _, err := range sendErrors {
 		if err != nil {
-			return "", fmt.Errorf("error publishing notification (service=%s): %w", service, err)
+			return "", fmt.Errorf("error publishing notification (msg=%s) (service=%s): %w", data.Message, service, err)
 		}
 	}