Complete Guide to Amazon SNS: Scalable Notification Service#

1
# Basic SNS Topic
2
NotificationTopic:
3
  Type: AWS::SNS::Topic
4
  Properties:
5
    TopicName: MyApplicationNotifications
6
    DisplayName: My Application Notifications
7
    KmsMasterKeyId: !Ref SNSKMSKey
8
    Tags:
9
      - Key: Application
10
        Value: MyApp
11
      - Key: Environment
12
        Value: Production
13

14
# Email subscription
15
EmailSubscription:
16
  Type: AWS::SNS::Subscription
17
  Properties:
18
    TopicArn: !Ref NotificationTopic
19
    Protocol: email
20
    Endpoint: admin@example.com
21
    FilterPolicy:
22
      event_type:
23
        - error
24
        - critical
25

26
# Lambda subscription
27
LambdaSubscription:
28
  Type: AWS::SNS::Subscription
29
  Properties:
30
    TopicArn: !Ref NotificationTopic
31
    Protocol: lambda
32
    Endpoint: !GetAtt NotificationProcessor.Arn
33
    FilterPolicy:
34
      source:
35
        - application
36
        - system

1
import boto3
2
import json
3
from datetime import datetime
4

5
sns = boto3.client('sns')
6

7
def publish_simple_message(topic_arn, message, subject=None):
8
    """
9
    Publish a simple text message
10
    """
11
    response = sns.publish(
12
        TopicArn=topic_arn,
13
        Message=message,
14
        Subject=subject
15
    )
16
    return response['MessageId']
17

18
def publish_structured_message(topic_arn, message_data):
19
    """
20
    Publish a structured message with different formats for different protocols
21
    """
22
    message = {
23
        'default': json.dumps(message_data),
24
        'email': f"Alert: {message_data.get('title', 'Notification')}\n\n{message_data.get('description', '')}",
25
        'sms': f"Alert: {message_data.get('title', '')[:50]}",
26
        'lambda': json.dumps(message_data),
27
        'sqs': json.dumps(message_data)
28
    }
29

30
    response = sns.publish(
31
        TopicArn=topic_arn,
32
        Message=json.dumps(message),
33
        MessageStructure='json',
34
        MessageAttributes={
35
            'event_type': {
36
                'DataType': 'String',
37
                'StringValue': message_data.get('event_type', 'info')
38
            },
39
            'source': {
40
                'DataType': 'String',
41
                'StringValue': message_data.get('source', 'application')
42
            },
43
            'timestamp': {
44
                'DataType': 'String',
45
                'StringValue': datetime.utcnow().isoformat()
46
            }
47
        }
48
    )
49
    return response['MessageId']
50

51
# Example usage
52
topic_arn = 'arn:aws:sns:us-east-1:123456789012:MyApplicationNotifications'
53

54
# Simple message
55
message_id = publish_simple_message(
56
    topic_arn,
57
    "System maintenance completed successfully",
58
    "Maintenance Update"
59
)
60

61
# Structured message
62
message_id = publish_structured_message(topic_arn, {
63
    'title': 'High CPU Usage Alert',
64
    'description': 'EC2 instance i-1234567890abcdef0 has exceeded 90% CPU usage',
65
    'event_type': 'warning',
66
    'source': 'monitoring',
67
    'instance_id': 'i-1234567890abcdef0',
68
    'cpu_usage': 92.5
69
})

1
# Subscription with message filtering
2
FilteredSubscription:
3
  Type: AWS::SNS::Subscription
4
  Properties:
5
    TopicArn: !Ref NotificationTopic
6
    Protocol: sqs
7
    Endpoint: !GetAtt CriticalAlertsQueue.Arn
8
    FilterPolicy:
9
      event_type:
10
        - error
11
        - critical
12
      source:
13
        - application
14
        - database
15
      severity:
16
        - high
17
        - critical
18
      region:
19
        - us-east-1
20
        - us-west-2
21

22
# Numeric filtering
23
NumericFilterSubscription:
24
  Type: AWS::SNS::Subscription
25
  Properties:
26
    TopicArn: !Ref MetricsTopic
27
    Protocol: lambda
28
    Endpoint: !GetAtt AlertProcessor.Arn
29
    FilterPolicy:
30
      cpu_usage:
31
        - ">": 80
32
      memory_usage:
33
        - ">=": 90
34
      error_rate:
35
        - "<": 5

1
# FIFO SNS Topic for ordered message delivery
2
FIFOTopic:
3
  Type: AWS::SNS::Topic
4
  Properties:
5
    TopicName: OrderProcessing.fifo
6
    FifoTopic: true
7
    ContentBasedDeduplication: true
8
    KmsMasterKeyId: !Ref SNSKMSKey
9

10
# FIFO SQS subscription
11
FIFOSubscription:
12
  Type: AWS::SNS::Subscription
13
  Properties:
14
    TopicArn: !Ref FIFOTopic
15
    Protocol: sqs
16
    Endpoint: !GetAtt OrderQueue.Arn

1
# Publishing to FIFO topic
2
def publish_to_fifo_topic(topic_arn, message, message_group_id, deduplication_id=None):
3
    """
4
    Publish message to FIFO topic with ordering
5
    """
6
    params = {
7
        'TopicArn': topic_arn,
8
        'Message': json.dumps(message),
9
        'MessageGroupId': message_group_id
10
    }
11

12
    # Add deduplication ID if not using content-based deduplication
13
    if deduplication_id:
14
        params['MessageDeduplicationId'] = deduplication_id
15

16
    response = sns.publish(**params)
17
    return response['MessageId']
18

19
# Example: Order processing messages
20
order_data = {
21
    'order_id': 'ORD-123456',
22
    'customer_id': 'CUST-789',
23
    'action': 'created',
24
    'timestamp': datetime.utcnow().isoformat()
25
}
26

27
message_id = publish_to_fifo_topic(
28
    'arn:aws:sns:us-east-1:123456789012:OrderProcessing.fifo',
29
    order_data,
30
    message_group_id=f"customer-{order_data['customer_id']}",
31
    deduplication_id=f"{order_data['order_id']}-{order_data['action']}"
32
)

1
# Platform application for mobile push
2
iOSPlatformApplication:
3
  Type: AWS::SNS::PlatformApplication
4
  Properties:
5
    PlatformApplicationName: MyApp-iOS
6
    Platform: APNS
7
    Attributes:
8
      PlatformCredential: !Ref APNSPrivateKey
9
      PlatformPrincipal: !Ref APNSCertificate
10

11
AndroidPlatformApplication:
12
  Type: AWS::SNS::PlatformApplication
13
  Properties:
14
    PlatformApplicationName: MyApp-Android
15
    Platform: GCM
16
    Attributes:
17
      PlatformCredential: !Ref FCMServerKey

1
# Mobile push notification management
2
class MobilePushManager:
3
    def __init__(self, region='us-east-1'):
4
        self.sns = boto3.client('sns', region_name=region)
5

6
    def create_platform_endpoint(self, platform_application_arn, token, user_data=None):
7
        """
8
        Create platform endpoint for device
9
        """
10
        attributes = {}
11
        if user_data:
12
            attributes['CustomUserData'] = json.dumps(user_data)
13

14
        response = self.sns.create_platform_endpoint(
15
            PlatformApplicationArn=platform_application_arn,
16
            Token=token,
17
            Attributes=attributes
18
        )
19
        return response['EndpointArn']
20

21
    def send_push_notification(self, endpoint_arn, message, badge_count=None, sound=None):
22
        """
23
        Send push notification to specific device
24
        """
25
        # Construct platform-specific payload
26
        ios_payload = {
27
            'aps': {
28
                'alert': message,
29
                'sound': sound or 'default'
30
            }
31
        }
32

33
        if badge_count is not None:
34
            ios_payload['aps']['badge'] = badge_count
35

36
        android_payload = {
37
            'data': {
38
                'message': message,
39
                'timestamp': datetime.utcnow().isoformat()
40
            }
41
        }
42

43
        # Create message structure
44
        message_structure = {
45
            'default': message,
46
            'APNS': json.dumps(ios_payload),
47
            'GCM': json.dumps(android_payload)
48
        }
49

50
        response = self.sns.publish(
51
            TargetArn=endpoint_arn,
52
            Message=json.dumps(message_structure),
53
            MessageStructure='json'
54
        )
55

56
        return response['MessageId']
57

58
    def send_bulk_notification(self, platform_application_arn, message, tokens):
59
        """
60
        Send notification to multiple devices
61
        """
62
        message_ids = []
63

64
        for token in tokens:
65
            try:
66
                endpoint_arn = self.create_platform_endpoint(platform_application_arn, token)
67
                message_id = self.send_push_notification(endpoint_arn, message)
68
                message_ids.append(message_id)
69
            except Exception as e:
70
                print(f"Failed to send notification to token {token}: {e}")
71

72
        return message_ids
73

74
# Usage example
75
push_manager = MobilePushManager()
76

77
# Send notification
78
endpoint_arn = push_manager.create_platform_endpoint(
79
    'arn:aws:sns:us-east-1:123456789012:app/APNS/MyApp-iOS',
80
    'device-token-here',
81
    user_data={'user_id': '12345', 'preferences': {'notifications': True}}
82
)
83

84
message_id = push_manager.send_push_notification(
85
    endpoint_arn,
86
    'You have a new message!',
87
    badge_count=1,
88
    sound='notification.wav'
89
)

1
# Delivery status logging
2
DeliveryStatusLogging:
3
  Type: AWS::SNS::Topic
4
  Properties:
5
    TopicName: MyTopicWithLogging
6
    DeliveryStatusLogging:
7
      - Protocol: lambda
8
        SuccessFeedbackRoleArn: !GetAtt SNSLoggingRole.Arn
9
        SuccessFeedbackSampleRate: 100
10
        FailureFeedbackRoleArn: !GetAtt SNSLoggingRole.Arn
11
      - Protocol: sqs
12
        SuccessFeedbackRoleArn: !GetAtt SNSLoggingRole.Arn
13
        SuccessFeedbackSampleRate: 50
14
        FailureFeedbackRoleArn: !GetAtt SNSLoggingRole.Arn
15

16
SNSLoggingRole:
17
  Type: AWS::IAM::Role
18
  Properties:
19
    AssumeRolePolicyDocument:
20
      Version: '2012-10-17'
21
      Statement:
22
        - Effect: Allow
23
          Principal:
24
            Service: sns.amazonaws.com
25
          Action: sts:AssumeRole
26
    Policies:
27
      - PolicyName: SNSLogsAccess
28
        PolicyDocument:
29
          Version: '2012-10-17'
30
          Statement:
31
            - Effect: Allow
32
              Action:
33
                - logs:CreateLogGroup
34
                - logs:CreateLogStream
35
                - logs:PutLogEvents
36
              Resource: '*'

1
# Fanout pattern with multiple SQS queues
2
PrimaryTopic:
3
  Type: AWS::SNS::Topic
4
  Properties:
5
    TopicName: OrderEvents
6

7
# Multiple processing queues
8
OrderProcessingQueue:
9
  Type: AWS::SQS::Queue
10
  Properties:
11
    QueueName: OrderProcessing
12
    MessageRetentionPeriod: 1209600  # 14 days
13
    VisibilityTimeoutSeconds: 300
14

15
EmailNotificationQueue:
16
  Type: AWS::SQS::Queue
17
  Properties:
18
    QueueName: EmailNotifications
19
    MessageRetentionPeriod: 345600   # 4 days
20

21
InventoryUpdateQueue:
22
  Type: AWS::SQS::Queue
23
  Properties:
24
    QueueName: InventoryUpdates
25
    MessageRetentionPeriod: 1209600
26

27
# Subscriptions
28
OrderProcessingSubscription:
29
  Type: AWS::SNS::Subscription
30
  Properties:
31
    TopicArn: !Ref PrimaryTopic
32
    Protocol: sqs
33
    Endpoint: !GetAtt OrderProcessingQueue.Arn
34

35
EmailSubscription:
36
  Type: AWS::SNS::Subscription
37
  Properties:
38
    TopicArn: !Ref PrimaryTopic
39
    Protocol: sqs
40
    Endpoint: !GetAtt EmailNotificationQueue.Arn
41
    FilterPolicy:
42
      event_type:
43
        - order_created
44
        - order_shipped
45

46
InventorySubscription:
47
  Type: AWS::SNS::Subscription
48
  Properties:
49
    TopicArn: !Ref PrimaryTopic
50
    Protocol: sqs
51
    Endpoint: !GetAtt InventoryUpdateQueue.Arn
52
    FilterPolicy:
53
      event_type:
54
        - order_created
55
        - order_cancelled

1
# Lambda function processing SNS messages
2
import json
3
import boto3
4
from datetime import datetime
5

6
def lambda_handler(event, context):
7
    """
8
    Process SNS messages in Lambda
9
    """
10
    processed_messages = []
11

12
    for record in event['Records']:
13
        # Extract SNS message
14
        sns_message = record['Sns']
15
        message_id = sns_message['MessageId']
16
        topic_arn = sns_message['TopicArn']
17

18
        try:
19
            # Parse message body
20
            if sns_message.get('MessageStructure') == 'json':
21
                message_body = json.loads(sns_message['Message'])
22
                actual_message = json.loads(message_body.get('default', '{}'))
23
            else:
24
                actual_message = json.loads(sns_message['Message'])
25

26
            # Extract message attributes
27
            attributes = sns_message.get('MessageAttributes', {})
28
            event_type = attributes.get('event_type', {}).get('Value', 'unknown')
29

30
            # Process based on event type
31
            result = process_message(actual_message, event_type, message_id)
32
            processed_messages.append(result)
33

34
        except Exception as e:
35
            print(f"Error processing message {message_id}: {str(e)}")
36
            # Send to dead letter queue or error handling
37
            handle_processing_error(message_id, sns_message, str(e))
38

39
    return {
40
        'statusCode': 200,
41
        'body': json.dumps({
42
            'processed_count': len(processed_messages),
43
            'results': processed_messages
44
        })
45
    }
46

47
def process_message(message, event_type, message_id):
48
    """
49
    Process individual message based on type
50
    """
51
    if event_type == 'order_created':
52
        return process_order_creation(message)
53
    elif event_type == 'user_signup':
54
        return process_user_signup(message)
55
    elif event_type == 'payment_failed':
56
        return process_payment_failure(message)
57
    else:
58
        return process_generic_message(message)
59

60
def process_order_creation(message):
61
    """
62
    Process new order creation
63
    """
64
    order_id = message.get('order_id')
65
    customer_id = message.get('customer_id')
66

67
    # Send welcome email
68
    send_order_confirmation(order_id, customer_id)
69

70
    # Update analytics
71
    update_order_analytics(message)
72

73
    return {'action': 'order_processed', 'order_id': order_id}

1
# Cross-account SNS topic policy
2
CrossAccountTopicPolicy:
3
  Type: AWS::SNS::TopicPolicy
4
  Properties:
5
    Topics:
6
      - !Ref SharedTopic
7
    PolicyDocument:
8
      Version: '2012-10-17'
9
      Statement:
10
        - Sid: AllowCrossAccountPublish
11
          Effect: Allow
12
          Principal:
13
            AWS:
14
              - 'arn:aws:iam::ACCOUNT-B:root'
15
              - 'arn:aws:iam::ACCOUNT-C:root'
16
          Action:
17
            - sns:Publish
18
          Resource: !Ref SharedTopic
19
          Condition:
20
            StringEquals:
21
              'sns:Protocol':
22
                - sqs
23
                - lambda
24
        - Sid: AllowCrossAccountSubscribe
25
          Effect: Allow
26
          Principal:
27
            AWS:
28
              - 'arn:aws:iam::ACCOUNT-B:root'
29
          Action:
30
            - sns:Subscribe
31
            - sns:Receive
32
          Resource: !Ref SharedTopic

1
# SNS CloudWatch Alarms
2
FailedPublishAlarm:
3
  Type: AWS::CloudWatch::Alarm
4
  Properties:
5
    AlarmName: SNS-Failed-Publishes
6
    AlarmDescription: SNS messages failed to publish
7
    MetricName: NumberOfMessagesFailed
8
    Namespace: AWS/SNS
9
    Statistic: Sum
10
    Period: 300
11
    EvaluationPeriods: 1
12
    Threshold: 1
13
    ComparisonOperator: GreaterThanOrEqualToThreshold
14
    Dimensions:
15
      - Name: TopicName
16
        Value: !GetAtt NotificationTopic.TopicName
17

18
DeliveryDelayAlarm:
19
  Type: AWS::CloudWatch::Alarm
20
  Properties:
21
    AlarmName: SNS-High-Delivery-Delay
22
    MetricName: PublishSize
23
    Namespace: AWS/SNS
24
    Statistic: Average
25
    Period: 300
26
    EvaluationPeriods: 2
27
    Threshold: 10000  # 10KB
28
    ComparisonOperator: GreaterThanThreshold

1
# SNS message tracking and analytics
2
import boto3
3
from datetime import datetime, timedelta
4

5
class SNSMonitor:
6
    def __init__(self, region='us-east-1'):
7
        self.sns = boto3.client('sns', region_name=region)
8
        self.cloudwatch = boto3.client('cloudwatch', region_name=region)
9

10
    def get_topic_metrics(self, topic_arn, start_time=None, end_time=None):
11
        """
12
        Get comprehensive metrics for SNS topic
13
        """
14
        if not start_time:
15
            start_time = datetime.utcnow() - timedelta(hours=24)
16
        if not end_time:
17
            end_time = datetime.utcnow()
18

19
        metrics = {}
20

21
        # Published messages
22
        response = self.cloudwatch.get_metric_statistics(
23
            Namespace='AWS/SNS',
24
            MetricName='NumberOfMessagesPublished',
25
            Dimensions=[
26
                {'Name': 'TopicName', 'Value': topic_arn.split(':')[-1]}
27
            ],
28
            StartTime=start_time,
29
            EndTime=end_time,
30
            Period=3600,
31
            Statistics=['Sum']
32
        )
33

34
        metrics['published'] = sum(point['Sum'] for point in response['Datapoints'])
35

36
        # Failed messages
37
        response = self.cloudwatch.get_metric_statistics(
38
            Namespace='AWS/SNS',
39
            MetricName='NumberOfMessagesFailed',
40
            Dimensions=[
41
                {'Name': 'TopicName', 'Value': topic_arn.split(':')[-1]}
42
            ],
43
            StartTime=start_time,
44
            EndTime=end_time,
45
            Period=3600,
46
            Statistics=['Sum']
47
        )
48

49
        metrics['failed'] = sum(point['Sum'] for point in response['Datapoints'])
50

51
        # Delivery attempts
52
        protocols = ['sqs', 'lambda', 'email', 'http']
53
        metrics['delivery_attempts'] = {}
54

55
        for protocol in protocols:
56
            response = self.cloudwatch.get_metric_statistics(
57
                Namespace='AWS/SNS',
58
                MetricName='NumberOfNotificationsDelivered',
59
                Dimensions=[
60
                    {'Name': 'TopicName', 'Value': topic_arn.split(':')[-1]},
61
                    {'Name': 'Protocol', 'Value': protocol}
62
                ],
63
                StartTime=start_time,
64
                EndTime=end_time,
65
                Period=3600,
66
                Statistics=['Sum']
67
            )
68

69
            metrics['delivery_attempts'][protocol] = sum(
70
                point['Sum'] for point in response['Datapoints']
71
            )
72

73
        return metrics
74

75
    def analyze_subscription_health(self, topic_arn):
76
        """
77
        Analyze health of topic subscriptions
78
        """
79
        # Get all subscriptions
80
        response = self.sns.list_subscriptions_by_topic(TopicArn=topic_arn)
81
        subscriptions = response['Subscriptions']
82

83
        health_report = []
84

85
        for subscription in subscriptions:
86
            sub_arn = subscription['SubscriptionArn']
87
            protocol = subscription['Protocol']
88
            endpoint = subscription['Endpoint']
89

90
            # Get subscription attributes
91
            try:
92
                attrs = self.sns.get_subscription_attributes(
93
                    SubscriptionArn=sub_arn
94
                )['Attributes']
95

96
                health_report.append({
97
                    'subscription_arn': sub_arn,
98
                    'protocol': protocol,
99
                    'endpoint': endpoint,
100
                    'confirmed': attrs.get('ConfirmationWasAuthenticated', 'false') == 'true',
101
                    'filter_policy': attrs.get('FilterPolicy'),
102
                    'delivery_policy': attrs.get('DeliveryPolicy')
103
                })
104
            except Exception as e:
105
                health_report.append({
106
                    'subscription_arn': sub_arn,
107
                    'protocol': protocol,
108
                    'endpoint': endpoint,
109
                    'error': str(e)
110
                })
111

112
        return health_report

1
# SNS topic with fine-grained access control
2
SecureTopic:
3
  Type: AWS::SNS::Topic
4
  Properties:
5
    KmsMasterKeyId: !Ref SNSKMSKey
6

7
SNSAccessPolicy:
8
  Type: AWS::SNS::TopicPolicy
9
  Properties:
10
    Topics:
11
      - !Ref SecureTopic
12
    PolicyDocument:
13
      Version: '2012-10-17'
14
      Statement:
15
        - Sid: AllowPublishFromLambda
16
          Effect: Allow
17
          Principal:
18
            AWS: !GetAtt LambdaExecutionRole.Arn
19
          Action:
20
            - sns:Publish
21
          Resource: !Ref SecureTopic
22
          Condition:
23
            StringEquals:
24
              'aws:SourceAccount': !Ref 'AWS::AccountId'
25
        - Sid: AllowSubscribeFromSameAccount
26
          Effect: Allow
27
          Principal:
28
            AWS: !Sub 'arn:aws:iam::${AWS::AccountId}:root'
29
          Action:
30
            - sns:Subscribe
31
            - sns:Receive
32
          Resource: !Ref SecureTopic
33
        - Sid: DenyUnencryptedMessages
34
          Effect: Deny
35
          Principal: '*'
36
          Action: sns:Publish
37
          Resource: !Ref SecureTopic
38
          Condition:
39
            Bool:
40
              'aws:SecureTransport': false

1
# KMS key for SNS encryption
2
SNSKMSKey:
3
  Type: AWS::KMS::Key
4
  Properties:
5
    Description: KMS key for SNS encryption
6
    KeyPolicy:
7
      Statement:
8
        - Effect: Allow
9
          Principal:
10
            AWS: !Sub "arn:aws:iam::${AWS::AccountId}:root"
11
          Action: "kms:*"
12
          Resource: "*"
13
        - Effect: Allow
14
          Principal:
15
            Service: sns.amazonaws.com
16
          Action:
17
            - kms:Decrypt
18
            - kms:GenerateDataKey
19
          Resource: "*"

1
def batch_publish_messages(topic_arn, messages, batch_size=10):
2
    """
3
    Batch multiple messages to reduce API calls
4
    """
5
    message_batches = [messages[i:i + batch_size] for i in range(0, len(messages), batch_size)]
6
    results = []
7

8
    for batch in message_batches:
9
        # For SNS, we can't truly batch, but we can optimize by reusing connections
10
        batch_results = []
11
        for message in batch:
12
            result = sns.publish(
13
                TopicArn=topic_arn,
14
                Message=json.dumps(message),
15
                MessageAttributes={
16
                    'batch_id': {
17
                        'DataType': 'String',
18
                        'StringValue': str(uuid.uuid4())
19
                    }
20
                }
21
            )
22
            batch_results.append(result['MessageId'])
23

24
        results.extend(batch_results)
25

26
    return results

1
def get_optimal_region_for_subscribers(subscribers):
2
    """
3
    Determine optimal region based on subscriber locations
4
    """
5
    region_counts = {}
6

7
    for subscriber in subscribers:
8
        # Parse ARNs or endpoints to determine regions
9
        if subscriber.startswith('arn:aws:'):
10
            region = subscriber.split(':')[3]
11
            region_counts[region] = region_counts.get(region, 0) + 1
12

13
    # Return region with most subscribers
14
    return max(region_counts.items(), key=lambda x: x[1])[0] if region_counts else 'us-east-1'