Implement notifications based on alert providers and severity

pkg/apis/flagger/v1beta1/canary.go

@@ -238,13 +238,22 @@ type CanaryThresholdRange struct {
 	Max *float64 `json:"max,omitempty"`
 }
 
+// AlertSeverity defines alert filtering based on severity levels
+type AlertSeverity string
+
+const (
+	SeverityInfo  AlertSeverity = "info"
+	SeverityWarn  AlertSeverity = "warn"
+	SeverityError AlertSeverity = "error"
+)
+
 // CanaryAlert defines an alert for this canary
 type CanaryAlert struct {
 	// Name of the alert
 	Name string `json:"name"`
 
 	// Severity level: info, warn, error (default info)
-	Severity string `json:"severity,omitempty"`
+	Severity AlertSeverity `json:"severity,omitempty"`
 
 	// Alert provider reference
 	ProviderRef CrossNamespaceObjectReference `json:"providerRef"`

pkg/controller/controller.go

@@ -261,94 +261,6 @@ func checkCustomResourceType(obj interface{}, logger *zap.SugaredLogger) (flagge
 	return *roll, true
 }
 
-func (c *Controller) sendEventToWebhook(r *flaggerv1.Canary, eventtype, template string, args []interface{}) {
-	webhookOverride := false
-	if len(r.Spec.CanaryAnalysis.Webhooks) > 0 {
-		for _, canaryWebhook := range r.Spec.CanaryAnalysis.Webhooks {
-			if canaryWebhook.Type == flaggerv1.EventHook {
-				webhookOverride = true
-				err := CallEventWebhook(r, canaryWebhook.URL, fmt.Sprintf(template, args...), eventtype)
-				if err != nil {
-					c.logger.With("canary", fmt.Sprintf("%s.%s", r.Name, r.Namespace)).Errorf("error sending event to webhook: %s", err)
-				}
-			}
-		}
-	}
-
-	if c.eventWebhook != "" && !webhookOverride {
-		err := CallEventWebhook(r, c.eventWebhook, fmt.Sprintf(template, args...), eventtype)
-		if err != nil {
-			c.logger.With("canary", fmt.Sprintf("%s.%s", r.Name, r.Namespace)).Errorf("error sending event to webhook: %s", err)
-		}
-	}
-}
-
-func (c *Controller) recordEventInfof(r *flaggerv1.Canary, template string, args ...interface{}) {
-	c.logger.With("canary", fmt.Sprintf("%s.%s", r.Name, r.Namespace)).Infof(template, args...)
-	c.eventRecorder.Event(r, corev1.EventTypeNormal, "Synced", fmt.Sprintf(template, args...))
-	c.sendEventToWebhook(r, corev1.EventTypeNormal, template, args)
-}
-
-func (c *Controller) recordEventErrorf(r *flaggerv1.Canary, template string, args ...interface{}) {
-	c.logger.With("canary", fmt.Sprintf("%s.%s", r.Name, r.Namespace)).Errorf(template, args...)
-	c.eventRecorder.Event(r, corev1.EventTypeWarning, "Synced", fmt.Sprintf(template, args...))
-	c.sendEventToWebhook(r, corev1.EventTypeWarning, template, args)
-}
-
-func (c *Controller) recordEventWarningf(r *flaggerv1.Canary, template string, args ...interface{}) {
-	c.logger.With("canary", fmt.Sprintf("%s.%s", r.Name, r.Namespace)).Infof(template, args...)
-	c.eventRecorder.Event(r, corev1.EventTypeWarning, "Synced", fmt.Sprintf(template, args...))
-	c.sendEventToWebhook(r, corev1.EventTypeWarning, template, args)
-}
-
-func (c *Controller) sendNotification(cd *flaggerv1.Canary, message string, metadata bool, warn bool) {
-	if c.notifier == nil {
-		return
-	}
-
-	var fields []notifier.Field
-
-	if metadata {
-		fields = append(fields,
-			notifier.Field{
-				Name:  "Target",
-				Value: fmt.Sprintf("%s/%s.%s", cd.Spec.TargetRef.Kind, cd.Spec.TargetRef.Name, cd.Namespace),
-			},
-			notifier.Field{
-				Name:  "Failed checks threshold",
-				Value: fmt.Sprintf("%v", cd.Spec.CanaryAnalysis.Threshold),
-			},
-			notifier.Field{
-				Name:  "Progress deadline",
-				Value: fmt.Sprintf("%vs", cd.GetProgressDeadlineSeconds()),
-			},
-		)
-
-		if cd.Spec.CanaryAnalysis.StepWeight > 0 {
-			fields = append(fields, notifier.Field{
-				Name: "Traffic routing",
-				Value: fmt.Sprintf("Weight step: %v max: %v",
-					cd.Spec.CanaryAnalysis.StepWeight,
-					cd.Spec.CanaryAnalysis.MaxWeight),
-			})
-		} else if len(cd.Spec.CanaryAnalysis.Match) > 0 {
-			fields = append(fields, notifier.Field{
-				Name:  "Traffic routing",
-				Value: "A/B Testing",
-			})
-		} else if cd.Spec.CanaryAnalysis.Iterations > 0 {
-			fields = append(fields, notifier.Field{
-				Name:  "Traffic routing",
-				Value: "Blue/Green",
-			})
-		}
-	}
-	err := c.notifier.Post(cd.Name, cd.Namespace, message, fields, warn)
-	if err != nil {
-		c.logger.Error(err)
-	}
-}
-
 func int32p(i int32) *int32 {
 	return &i
 }

pkg/controller/events.go

@@ -0,0 +1,191 @@
+package controller
+
+import (
+	"fmt"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	corev1 "k8s.io/api/core/v1"
+
+	flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1beta1"
+	"github.com/weaveworks/flagger/pkg/notifier"
+)
+
+func (c *Controller) recordEventInfof(r *flaggerv1.Canary, template string, args ...interface{}) {
+	c.logger.With("canary", fmt.Sprintf("%s.%s", r.Name, r.Namespace)).Infof(template, args...)
+	c.eventRecorder.Event(r, corev1.EventTypeNormal, "Synced", fmt.Sprintf(template, args...))
+	c.sendEventToWebhook(r, corev1.EventTypeNormal, template, args)
+}
+
+func (c *Controller) recordEventErrorf(r *flaggerv1.Canary, template string, args ...interface{}) {
+	c.logger.With("canary", fmt.Sprintf("%s.%s", r.Name, r.Namespace)).Errorf(template, args...)
+	c.eventRecorder.Event(r, corev1.EventTypeWarning, "Synced", fmt.Sprintf(template, args...))
+	c.sendEventToWebhook(r, corev1.EventTypeWarning, template, args)
+}
+
+func (c *Controller) recordEventWarningf(r *flaggerv1.Canary, template string, args ...interface{}) {
+	c.logger.With("canary", fmt.Sprintf("%s.%s", r.Name, r.Namespace)).Infof(template, args...)
+	c.eventRecorder.Event(r, corev1.EventTypeWarning, "Synced", fmt.Sprintf(template, args...))
+	c.sendEventToWebhook(r, corev1.EventTypeWarning, template, args)
+}
+
+func (c *Controller) sendEventToWebhook(r *flaggerv1.Canary, eventType, template string, args []interface{}) {
+	webhookOverride := false
+	if len(r.Spec.CanaryAnalysis.Webhooks) > 0 {
+		for _, canaryWebhook := range r.Spec.CanaryAnalysis.Webhooks {
+			if canaryWebhook.Type == flaggerv1.EventHook {
+				webhookOverride = true
+				err := CallEventWebhook(r, canaryWebhook.URL, fmt.Sprintf(template, args...), eventType)
+				if err != nil {
+					c.logger.With("canary", fmt.Sprintf("%s.%s", r.Name, r.Namespace)).Errorf("error sending event to webhook: %s", err)
+				}
+			}
+		}
+	}
+
+	if c.eventWebhook != "" && !webhookOverride {
+		err := CallEventWebhook(r, c.eventWebhook, fmt.Sprintf(template, args...), eventType)
+		if err != nil {
+			c.logger.With("canary", fmt.Sprintf("%s.%s", r.Name, r.Namespace)).Errorf("error sending event to webhook: %s", err)
+		}
+	}
+}
+
+func (c *Controller) alert(canary *flaggerv1.Canary, message string, metadata bool, severity flaggerv1.AlertSeverity) {
+	if c.notifier == nil && len(canary.Spec.CanaryAnalysis.Alerts) == 0 {
+		return
+	}
+
+	var fields []notifier.Field
+	if metadata {
+		fields = alertMetadata(canary)
+	}
+
+	// send alert with the global notifier
+	if len(canary.Spec.CanaryAnalysis.Alerts) == 0 {
+		err := c.notifier.Post(canary.Name, canary.Namespace, message, fields, string(severity))
+		if err != nil {
+			c.logger.With("canary", fmt.Sprintf("%s.%s", canary.Name, canary.Namespace)).
+				Errorf("alert can't be sent: %v", err)
+			return
+		}
+		return
+	}
+
+	// send canary alerts
+	for _, alert := range canary.Spec.CanaryAnalysis.Alerts {
+		// determine if alert should be sent based on severity level
+		shouldAlert := false
+		if alert.Severity == flaggerv1.SeverityInfo {
+			shouldAlert = true
+		} else {
+			if severity == alert.Severity {
+				shouldAlert = true
+			}
+			if severity == flaggerv1.SeverityWarn && alert.Severity == flaggerv1.SeverityError {
+				shouldAlert = true
+			}
+		}
+		if !shouldAlert {
+			continue
+		}
+
+		// determine alert provider namespace
+		providerNamespace := canary.GetNamespace()
+		if alert.ProviderRef.Namespace != "" {
+			providerNamespace = alert.ProviderRef.Namespace
+		}
+
+		// find alert provider
+		provider, err := c.flaggerInformers.AlertInformer.Lister().AlertProviders(providerNamespace).Get(alert.ProviderRef.Name)
+		if err != nil {
+			c.logger.With("canary", fmt.Sprintf("%s.%s", canary.Name, canary.Namespace)).
+				Errorf("alert provider %s.%s error: %v", alert.ProviderRef.Name, providerNamespace, err)
+			continue
+		}
+
+		// set hook URL address
+		url := provider.Spec.Address
+
+		// extract address from secret
+		if provider.Spec.SecretRef != nil {
+			secret, err := c.kubeClient.CoreV1().Secrets(providerNamespace).Get(provider.Spec.SecretRef.Name, metav1.GetOptions{})
+			if err != nil {
+				c.logger.With("canary", fmt.Sprintf("%s.%s", canary.Name, canary.Namespace)).
+					Errorf("alert provider %s.%s secretRef error: %v", alert.ProviderRef.Name, providerNamespace, err)
+				continue
+			}
+			if address, ok := secret.Data["address"]; ok {
+				url = string(address)
+			} else {
+				c.logger.With("canary", fmt.Sprintf("%s.%s", canary.Name, canary.Namespace)).
+					Errorf("alert provider %s.%s secret does not contain an address", alert.ProviderRef.Name, providerNamespace)
+				continue
+			}
+		}
+
+		// set defaults
+		username := "flagger"
+		if provider.Spec.Username != "" {
+			username = provider.Spec.Username
+		}
+		channel := "general"
+		if provider.Spec.Channel != "" {
+			channel = provider.Spec.Channel
+		}
+
+		// create notifier based on provider type
+		f := notifier.NewFactory(url, username, channel)
+		n, err := f.Notifier(provider.Spec.Type)
+		if err != nil {
+			c.logger.With("canary", fmt.Sprintf("%s.%s", canary.Name, canary.Namespace)).
+				Errorf("alert provider %s.%s error: %v", alert.ProviderRef.Name, providerNamespace, err)
+			continue
+		}
+
+		// send alert
+		err = n.Post(canary.Name, canary.Namespace, message, fields, string(severity))
+		if err != nil {
+			c.logger.With("canary", fmt.Sprintf("%s.%s", canary.Name, canary.Namespace)).
+				Errorf("alert provider $s.%s send error: %v", alert.ProviderRef.Name, providerNamespace, err)
+		}
+
+	}
+}
+
+func alertMetadata(canary *flaggerv1.Canary) []notifier.Field {
+	var fields []notifier.Field
+	fields = append(fields,
+		notifier.Field{
+			Name:  "Target",
+			Value: fmt.Sprintf("%s/%s.%s", canary.Spec.TargetRef.Kind, canary.Spec.TargetRef.Name, canary.Namespace),
+		},
+		notifier.Field{
+			Name:  "Failed checks threshold",
+			Value: fmt.Sprintf("%v", canary.Spec.CanaryAnalysis.Threshold),
+		},
+		notifier.Field{
+			Name:  "Progress deadline",
+			Value: fmt.Sprintf("%vs", canary.GetProgressDeadlineSeconds()),
+		},
+	)
+
+	if canary.Spec.CanaryAnalysis.StepWeight > 0 {
+		fields = append(fields, notifier.Field{
+			Name: "Traffic routing",
+			Value: fmt.Sprintf("Weight step: %v max: %v",
+				canary.Spec.CanaryAnalysis.StepWeight,
+				canary.Spec.CanaryAnalysis.MaxWeight),
+		})
+	} else if len(canary.Spec.CanaryAnalysis.Match) > 0 {
+		fields = append(fields, notifier.Field{
+			Name:  "Traffic routing",
+			Value: "A/B Testing",
+		})
+	} else if canary.Spec.CanaryAnalysis.Iterations > 0 {
+		fields = append(fields, notifier.Field{
+			Name:  "Traffic routing",
+			Value: "Blue/Green",
+		})
+	}
+	return fields
+}

pkg/controller/scheduler.go

@@ -230,7 +230,7 @@ func (c *Controller) advanceCanary(name string, namespace string, skipLivenessCh
 		cd.Status.Phase == flaggerv1.CanaryPhaseWaiting {
 		if ok := c.runRollbackHooks(cd, cd.Status.Phase); ok {
 			c.recordEventWarningf(cd, "Rolling back %s.%s manual webhook invoked", cd.Name, cd.Namespace)
-			c.sendNotification(cd, "Rolling back manual webhook invoked", false, true)
+			c.alert(cd, "Rolling back manual webhook invoked", false, flaggerv1.SeverityWarn)
 			c.rollback(cd, canaryController, meshRouter)
 			return
 		}
@@ -271,8 +271,8 @@ func (c *Controller) advanceCanary(name string, namespace string, skipLivenessCh
 		c.recorder.SetStatus(cd, flaggerv1.CanaryPhaseSucceeded)
 		c.runPostRolloutHooks(cd, flaggerv1.CanaryPhaseSucceeded)
 		c.recordEventInfof(cd, "Promotion completed! Scaling down %s.%s", cd.Spec.TargetRef.Name, cd.Namespace)
-		c.sendNotification(cd, "Canary analysis completed successfully, promotion finished.",
-			false, false)
+		c.alert(cd, "Canary analysis completed successfully, promotion finished.",
+			false, flaggerv1.SeverityInfo)
 		return
 	}
 
@@ -282,8 +282,8 @@ func (c *Controller) advanceCanary(name string, namespace string, skipLivenessCh
 		if !retriable {
 			c.recordEventWarningf(cd, "Rolling back %s.%s progress deadline exceeded %v",
 				cd.Name, cd.Namespace, err)
-			c.sendNotification(cd, fmt.Sprintf("Progress deadline exceeded %v", err),
-				false, true)
+			c.alert(cd, fmt.Sprintf("Progress deadline exceeded %v", err),
+				false, flaggerv1.SeverityError)
 		}
 		c.rollback(cd, canaryController, meshRouter)
 		return
@@ -569,8 +569,8 @@ func (c *Controller) shouldSkipAnalysis(canary *flaggerv1.Canary, canaryControll
 	c.recorder.SetStatus(canary, flaggerv1.CanaryPhaseSucceeded)
 	c.recordEventInfof(canary, "Promotion completed! Canary analysis was skipped for %s.%s",
 		canary.Spec.TargetRef.Name, canary.Namespace)
-	c.sendNotification(canary, "Canary analysis was skipped, promotion finished.",
-		false, false)
+	c.alert(canary, "Canary analysis was skipped, promotion finished.",
+		false, flaggerv1.SeverityInfo)
 
 	return true
 }
@@ -617,8 +617,8 @@ func (c *Controller) checkCanaryStatus(canary *flaggerv1.Canary, canaryControlle
 		}
 		c.recorder.SetStatus(canary, flaggerv1.CanaryPhaseInitialized)
 		c.recordEventInfof(canary, "Initialization done! %s.%s", canary.Name, canary.Namespace)
-		c.sendNotification(canary, "New deployment detected, initialization completed.",
-			true, false)
+		c.alert(canary, "New deployment detected, initialization completed.",
+			true, flaggerv1.SeverityInfo)
 		return false
 	}
 
@@ -626,8 +626,8 @@ func (c *Controller) checkCanaryStatus(canary *flaggerv1.Canary, canaryControlle
 		canaryPhaseProgressing := canary.DeepCopy()
 		canaryPhaseProgressing.Status.Phase = flaggerv1.CanaryPhaseProgressing
 		c.recordEventInfof(canaryPhaseProgressing, "New revision detected! Scaling up %s.%s", canaryPhaseProgressing.Spec.TargetRef.Name, canaryPhaseProgressing.Namespace)
-		c.sendNotification(canaryPhaseProgressing, "New revision detected, starting canary analysis.",
-			true, false)
+		c.alert(canaryPhaseProgressing, "New revision detected, starting canary analysis.",
+			true, flaggerv1.SeverityInfo)
 
 		if err := canaryController.ScaleFromZero(canary); err != nil {
 			c.recordEventErrorf(canary, "%v", err)
@@ -666,7 +666,7 @@ func (c *Controller) runConfirmRolloutHooks(canary *flaggerv1.Canary, canaryCont
 					}
 					c.recordEventWarningf(canary, "Halt %s.%s advancement waiting for approval %s",
 						canary.Name, canary.Namespace, webhook.Name)
-					c.sendNotification(canary, "Canary is waiting for approval.", false, false)
+					c.alert(canary, "Canary is waiting for approval.", false, flaggerv1.SeverityWarn)
 				}
 				return false
 			} else {
@@ -691,7 +691,7 @@ func (c *Controller) runConfirmPromotionHooks(canary *flaggerv1.Canary) bool {
 			if err != nil {
 				c.recordEventWarningf(canary, "Halt %s.%s advancement waiting for promotion approval %s",
 					canary.Name, canary.Namespace, webhook.Name)
-				c.sendNotification(canary, "Canary promotion is waiting for approval.", false, false)
+				c.alert(canary, "Canary promotion is waiting for approval.", false, flaggerv1.SeverityWarn)
 				return false
 			} else {
 				c.recordEventInfof(canary, "Confirm-promotion check %s passed", webhook.Name)
@@ -1010,8 +1010,8 @@ func (c *Controller) rollback(canary *flaggerv1.Canary, canaryController canary.
 	if canary.Status.FailedChecks >= canary.Spec.CanaryAnalysis.Threshold {
 		c.recordEventWarningf(canary, "Rolling back %s.%s failed checks threshold reached %v",
 			canary.Name, canary.Namespace, canary.Status.FailedChecks)
-		c.sendNotification(canary, fmt.Sprintf("Failed checks threshold reached %v", canary.Status.FailedChecks),
-			false, true)
+		c.alert(canary, fmt.Sprintf("Failed checks threshold reached %v", canary.Status.FailedChecks),
+			false, flaggerv1.SeverityError)
 	}
 
 	// route all traffic back to primary

pkg/notifier/discord.go

@@ -45,7 +45,7 @@ func NewDiscord(hookURL string, username string, channel string) (*Discord, erro
 }
 
 // Post Discord message
-func (s *Discord) Post(workload string, namespace string, message string, fields []Field, warn bool) error {
+func (s *Discord) Post(workload string, namespace string, message string, fields []Field, severity string) error {
 	payload := SlackPayload{
 		Channel:   s.Channel,
 		Username:  s.Username,
@@ -53,7 +53,7 @@ func (s *Discord) Post(workload string, namespace string, message string, fields
 	}
 
 	color := "good"
-	if warn {
+	if severity == "error" {
 		color = "danger"
 	}
 

pkg/notifier/discord_test.go

@@ -42,7 +42,7 @@ func TestDiscord_Post(t *testing.T) {
 		t.Error("Invalid Discord URL, expected to have /slack prefix")
 	}
 
-	err = discord.Post("podinfo", "test", "test", fields, true)
+	err = discord.Post("podinfo", "test", "test", fields, "warn")
 	if err != nil {
 		t.Fatal(err)
 	}

pkg/notifier/factory.go

@@ -1,5 +1,7 @@
 package notifier
 
+import "fmt"
+
 type Factory struct {
 	URL      string
 	Username string
@@ -24,5 +26,5 @@ func (f Factory) Notifier(provider string) (Interface, error) {
 		return NewMSTeams(f.URL)
 	}
 
-	return nil, nil
+	return nil, fmt.Errorf("provider %s not supported", provider)
 }

pkg/notifier/notifier.go

@@ -1,7 +1,7 @@
 package notifier
 
 type Interface interface {
-	Post(workload string, namespace string, message string, fields []Field, warn bool) error
+	Post(workload string, namespace string, message string, fields []Field, severity string) error
 }
 
 type Field struct {

pkg/notifier/slack.go

@@ -61,7 +61,7 @@ func NewSlack(hookURL string, username string, channel string) (*Slack, error) {
 }
 
 // Post Slack message
-func (s *Slack) Post(workload string, namespace string, message string, fields []Field, warn bool) error {
+func (s *Slack) Post(workload string, namespace string, message string, fields []Field, severity string) error {
 	payload := SlackPayload{
 		Channel:   s.Channel,
 		Username:  s.Username,
@@ -69,7 +69,7 @@ func (s *Slack) Post(workload string, namespace string, message string, fields [
 	}
 
 	color := "good"
-	if warn {
+	if severity == "error" {
 		color = "danger"
 	}
 

pkg/notifier/slack_test.go

@@ -37,7 +37,7 @@ func TestSlack_Post(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	err = slack.Post("podinfo", "test", "test", fields, true)
+	err = slack.Post("podinfo", "test", "test", fields, "error")
 	if err != nil {
 		t.Fatal(err)
 	}

pkg/notifier/teams.go

@@ -44,7 +44,7 @@ func NewMSTeams(hookURL string) (*MSTeams, error) {
 }
 
 // Post MS Teams message
-func (s *MSTeams) Post(workload string, namespace string, message string, fields []Field, warn bool) error {
+func (s *MSTeams) Post(workload string, namespace string, message string, fields []Field, severity string) error {
 	facts := make([]MSTeamsField, 0, len(fields))
 	for _, f := range fields {
 		facts = append(facts, MSTeamsField{f.Name, f.Value})
@@ -64,7 +64,7 @@ func (s *MSTeams) Post(workload string, namespace string, message string, fields
 		},
 	}
 
-	if warn {
+	if severity == "error" {
 		payload.ThemeColor = "FF0000"
 	}
 

pkg/notifier/teams_test.go

@@ -37,7 +37,7 @@ func TestTeams_Post(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	err = teams.Post("podinfo", "test", "test", fields, true)
+	err = teams.Post("podinfo", "test", "test", fields, "info")
 	if err != nil {
 		t.Fatal(err)
 	}