Add AWS SNS receiver

config/config_test.go

@@ -26,7 +26,7 @@ import (
 	commoncfg "github.com/prometheus/common/config"
 	"github.com/prometheus/common/model"
 	"github.com/stretchr/testify/require"
-	yaml "gopkg.in/yaml.v2"
+	"gopkg.in/yaml.v2"
 )
 
 func TestLoadEmptyString(t *testing.T) {
@@ -904,6 +904,20 @@ func TestSlackGlobalAPIURLFile(t *testing.T) {
 	}
 }
 
+func TestValidSNSConfig(t *testing.T) {
+	_, err := LoadFile("testdata/conf.sns-topic-arn.yml")
+	if err != nil {
+		t.Fatalf("Error parsing %s: %s", "testdata/conf.sns-topic-arn.yml\"", err)
+	}
+}
+
+func TestInvalidSNSConfig(t *testing.T) {
+	_, err := LoadFile("testdata/conf.sns-invalid.yml")
+	if err == nil {
+		t.Fatalf("expected error with missing fields on SNS config")
+	}
+}
+
 func TestUnmarshalHostPort(t *testing.T) {
 	for _, tc := range []struct {
 		in string

config/notifiers.go

@@ -135,7 +135,6 @@ var (
 		},
 		APIVersion:  "sns.default.api_version",
 		Message:     `{{ template "sns.default.message" . }}`,
-		IsFIFOTopic: false,
 	}
 )
 
@@ -590,11 +589,11 @@ func (c *PushoverConfig) UnmarshalYAML(unmarshal func(interface{}) error) error
 	return nil
 }
 
-// TODO: Move to common?
 
 // SigV4Config is the configuration for signing remote write requests with
 // AWS's SigV4 verification process. Empty values will be retrieved using the
 // AWS default credentials chain.
+// TODO: Move to common.
 type SigV4Config struct {
 	Region    string `yaml:"region,omitempty"`
 	AccessKey string `yaml:"access_key,omitempty"`
@@ -614,7 +613,6 @@ type SNSConfig struct {
 	TopicARN    string            `yaml:"topic_arn,omitempty" json:"topic_arn,omitempty"`
 	PhoneNumber string            `yaml:"phone_number,omitempty" json:"phone_number,omitempty"`
 	TargetARN   string            `yaml:"target_arn,omitempty" json:"target_arn,omitempty"`
-	IsFIFOTopic bool              `yaml:"is_fifo_topic,omitempty" json:"is_fifo_topic,omitempty"`
 	Subject     string            `yaml:"subject,omitempty" json:"subject,omitempty"`
 	Message     string            `yaml:"message,omitempty" json:"message,omitempty"`
 	Attributes  map[string]string `yaml:"attributes,omitempty" json:"attributes,omitempty"`

config/testdata/conf.sns-invalid.yml

@@ -0,0 +1,14 @@
+route:
+  receiver: 'sns-api-notifications'
+  group_by: [alertname]
+
+receivers:
+  - name: 'sns-api-notifications'
+    sns_configs:
+      - api_url: https://sns.us-east-2.amazonaws.com
+        sigv4:
+          region: us-east-2
+          access_key: access_key
+          secret_key: secret_ket
+        attributes:
+          severity: Sev2

config/testdata/conf.sns-test.yml → config/testdata/conf.sns-topic-arn.yml

@@ -7,7 +7,6 @@ receivers:
   sns_configs:
     - api_url: https://sns.us-east-2.amazonaws.com
       topic_arn: arn:aws:sns:us-east-2:123456789012:My-Topic
-      is_fifo_topic: true
       sigv4:
         region: us-east-2
         access_key: access_key

notify/sns/sns.go

@@ -17,9 +17,11 @@ import (
 	"context"
 	"fmt"
 	"net/http"
+	"strings"
 	"unicode/utf8"
 
 	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/awserr"
 	"github.com/aws/aws-sdk-go/aws/credentials"
 	"github.com/aws/aws-sdk-go/aws/credentials/stscreds"
 	"github.com/aws/aws-sdk-go/aws/session"
@@ -41,17 +43,38 @@ type Notifier struct {
 	retrier *notify.Retrier
 }
 
+// New returns a new SNS notification handler.
+func New(c *config.SNSConfig, t *template.Template, l log.Logger, httpOpts ...commoncfg.HTTPClientOption) (*Notifier, error) {
+	client, err := commoncfg.NewClientFromConfig(*c.HTTPConfig, "sns", append(httpOpts, commoncfg.WithHTTP2Disabled())...)
+	if err != nil {
+		return nil, err
+	}
+
+	return &Notifier{
+		conf:    c,
+		tmpl:    t,
+		logger:  l,
+		client:  client,
+		retrier: &notify.Retrier{},
+	}, nil
+}
+
 func (n Notifier) Notify(ctx context.Context, alert ...*types.Alert) (bool, error) {
-	credentials := credentials.NewStaticCredentials(n.conf.Sigv4.AccessKey, string(n.conf.Sigv4.SecretKey), "")
-	if n.conf.Sigv4.AccessKey == "" {
-		credentials = nil
+	var(
+		err error
+		data = notify.GetTemplateData(ctx, n.tmpl, alert, n.logger)
+		tmpl = notify.TmplText(n.tmpl, data, &err)
+		creds *credentials.Credentials = nil
+	)
+	if n.conf.Sigv4.AccessKey != "" && n.conf.Sigv4.SecretKey != "" {
+		creds = credentials.NewStaticCredentials(n.conf.Sigv4.AccessKey, string(n.conf.Sigv4.SecretKey), "")
 	}
 
 	sess, err := session.NewSessionWithOptions(session.Options{
 		Config: aws.Config{
 			Region:      aws.String(n.conf.Sigv4.Region),
-			Credentials: credentials,
-			Endpoint:    aws.String(n.conf.APIUrl),
+			Credentials: creds,
+			Endpoint:    aws.String(tmpl(n.conf.APIUrl)),
 		},
 		Profile: n.conf.Sigv4.Profile,
 	})
@@ -64,37 +87,44 @@ func (n Notifier) Notify(ctx context.Context, alert ...*types.Alert) (bool, erro
 		sess.Config.Credentials = stscreds.NewCredentials(sess, n.conf.Sigv4.RoleARN)
 	}
 
-	data := notify.GetTemplateData(ctx, n.tmpl, alert, n.logger)
-	tmpl := notify.TmplText(n.tmpl, data, &err)
-	message := tmpl(n.conf.Message)
-
-	client := sns.New(sess, &aws.Config{Credentials: credentials})
+	client := sns.New(sess, &aws.Config{Credentials: creds})
 	publishInput := &sns.PublishInput{}
 
 	if n.conf.TopicARN != "" {
-		publishInput.SetTopicArn(n.conf.TopicARN)
-		messageToSend, isTrunc, err := validateAndTruncateMessage(message)
+		publishInput.SetTopicArn(tmpl(n.conf.TopicARN))
+		messageToSend, isTrunc, err := validateAndTruncateMessage(tmpl(n.conf.Message))
 		if err != nil {
 			return false, err
 		}
 		if isTrunc {
 			n.conf.Attributes["truncated"] = "true"
 		}
+
+		// Deduplication key and Message Group ID are only added if it's a FIFO SNS Topic.
+		if isFIFOTopic(n.conf.TopicARN) {
+			key, err := notify.ExtractGroupKey(ctx)
+			if err != nil {
+				return false, err
+			}
+			publishInput.SetMessageDeduplicationId(key.Hash())
+			publishInput.SetMessageGroupId(key.Hash())
+		}
+
 		publishInput.SetMessage(messageToSend)
 	}
 	if n.conf.PhoneNumber != "" {
-		publishInput.SetPhoneNumber(n.conf.PhoneNumber)
+		publishInput.SetPhoneNumber(tmpl(n.conf.PhoneNumber))
 		// If SMS message is over 1600 chars, SNS will reject the message.
-		_, isTruncated := notify.Truncate(message, 1600)
+		_, isTruncated := notify.Truncate(tmpl(n.conf.Message), 1600)
 		if isTruncated {
 			return false, fmt.Errorf("SMS message exeeds length of 1600 charactors")
 		} else {
-			publishInput.SetMessage(message)
+			publishInput.SetMessage(tmpl(n.conf.Message))
 		}
 	}
 	if n.conf.TargetARN != "" {
-		publishInput.SetTargetArn(n.conf.TargetARN)
-		messageToSend, isTrunc, err := validateAndTruncateMessage(message)
+		publishInput.SetTargetArn(tmpl(n.conf.TargetARN))
+		messageToSend, isTrunc, err := validateAndTruncateMessage(tmpl(n.conf.Message))
 		if err != nil {
 			return false, err
 		}
@@ -107,39 +137,35 @@ func (n Notifier) Notify(ctx context.Context, alert ...*types.Alert) (bool, erro
 	if len(n.conf.Attributes) > 0 {
 		attributes := map[string]*sns.MessageAttributeValue{}
 		for k, v := range n.conf.Attributes {
-			attributes[k] = &sns.MessageAttributeValue{DataType: aws.String("String"), StringValue: aws.String(v)}
+			attributes[tmpl(k)] = &sns.MessageAttributeValue{DataType: aws.String("String"), StringValue: aws.String(tmpl(v))}
 		}
 		publishInput.SetMessageAttributes(attributes)
 	}
 
 	if n.conf.Subject != "" {
-		publishInput.SetSubject(n.conf.Subject)
-	}
-
-	// Deduplication key is only added if it's a FIFO SNS Topic.
-	if n.conf.IsFIFOTopic {
-		key, err := notify.ExtractGroupKey(ctx)
-		if err != nil {
-			return false, err
-		}
-		publishInput.SetMessageDeduplicationId(key.Hash())
+		publishInput.SetSubject(tmpl(n.conf.Subject))
 	}
 
 	publishOutput, err := client.Publish(publishInput)
 	if err != nil {
-		// AWS Response is bad, probably a config issue.
-		return false, err
+		return n.retrier.Check(err.(awserr.RequestFailure).StatusCode(), strings.NewReader(err.(awserr.RequestFailure).Message()))
 	}
 
 	err = n.logger.Log(publishOutput.String())
 	if err != nil {
 		return false, err
 	}
 
-	// Response is good and does not need to be retried.
 	return false, nil
 }
 
+func isFIFOTopic(topicARN string) bool {
+	if topicARN[len(topicARN)-5:] == ".fifo" {
+		return true
+	}
+	return false
+}
+
 func validateAndTruncateMessage(message string) (string, bool, error) {
 	if utf8.ValidString(message) {
 		// if the message is larger than 256KB we have to truncate.
@@ -152,20 +178,3 @@ func validateAndTruncateMessage(message string) (string, bool, error) {
 	}
 	return "", false, fmt.Errorf("non utf8 encoded message string")
 }
-
-// New returns a new SNS notification handler.
-func New(c *config.SNSConfig, t *template.Template, l log.Logger, httpOpts ...commoncfg.HTTPClientOption) (*Notifier, error) {
-
-	client, err := commoncfg.NewClientFromConfig(*c.HTTPConfig, "sns", append(httpOpts, commoncfg.WithHTTP2Disabled())...)
-	if err != nil {
-		return nil, err
-	}
-
-	return &Notifier{
-		conf:    c,
-		tmpl:    t,
-		logger:  l,
-		client:  client,
-		retrier: &notify.Retrier{},
-	}, nil
-}

notify/sns/sns_test.go

@@ -15,51 +15,36 @@ package sns
 
 import (
 	"testing"
-	"time"
 
-	"github.com/go-kit/kit/log"
-	"github.com/prometheus/alertmanager/config"
-	"github.com/prometheus/alertmanager/notify/test"
-	"github.com/prometheus/alertmanager/types"
-	commoncfg "github.com/prometheus/common/config"
-	"github.com/prometheus/common/model"
 	"github.com/stretchr/testify/require"
 )
 
-func TestNotifier_Notify(t *testing.T) {
-	ctx, _, fn := test.GetContextWithCancelingURL()
-	defer fn()
-	attrTest := map[string]string{}
-	attrTest["key"] = "testVal"
-	// These are fake values
-	notifier, err := New(
-		&config.SNSConfig{
-			HTTPConfig: &commoncfg.HTTPClientConfig{},
-			Message:    `{{ template "sns.default.message" . }}`,
-			TopicARN:   "arn:aws:sns:us-east-2:123456789012:My-Topic",
-			Sigv4: config.SigV4Config{
-				Region:    "us-east-2",
-				AccessKey: "access_key",
-				SecretKey: "secret_key",
-			},
-			Attributes: attrTest,
-		},
-		test.CreateTmpl(t),
-		log.NewNopLogger(),
-	)
+func TestIsFIFO(t *testing.T) {
+	require.True(t, isFIFOTopic("arn:aws:sns:us-east-2:624413706616:snsTestTopic.fifo"))
+	require.False(t, isFIFOTopic("arn:aws:sns:us-east-2:624413706616:snsTestTopic"))
+}
+
+func TestValidateAndTruncateMessage(t *testing.T) {
+	sBuff := make([]byte, 257*1024, 257*1024)
+	for i := range sBuff {
+		sBuff[i] = byte(33)
+	}
+	truncatedMessage, isTruncated, err := validateAndTruncateMessage(string(sBuff))
+	require.True(t, isTruncated)
 	require.NoError(t, err)
+	require.NotEqual(t, sBuff, truncatedMessage)
+	require.Equal(t, len(truncatedMessage), 256*1024)
 
-	ok, err := notifier.Notify(ctx, []*types.Alert{
-		&types.Alert{
-			Alert: model.Alert{
-				Labels: model.LabelSet{
-					"lbl1": "val1",
-				},
-				StartsAt: time.Now(),
-				EndsAt:   time.Now().Add(time.Hour),
-			},
-		},
-	}...)
+	sBuff = make([]byte, 100, 100)
+	for i := range sBuff {
+		sBuff[i] = byte(33)
+	}
+	truncatedMessage, isTruncated, err = validateAndTruncateMessage(string(sBuff))
+	require.False(t, isTruncated)
 	require.NoError(t, err)
-	require.False(t, ok)
+	require.Equal(t, string(sBuff), truncatedMessage)
+
+	invalidUtf8String := "\xc3\x28"
+	_, _, err = validateAndTruncateMessage(invalidUtf8String)
+	require.Error(t, err)
 }