Complete Guide to Amazon SQS: Scalable Message Queuing#

1
# Standard SQS Queue
2
StandardQueue:
3
  Type: AWS::SQS::Queue
4
  Properties:
5
    QueueName: standard-queue
6
    MessageRetentionPeriod: 1209600  # 14 days
7
    VisibilityTimeoutSeconds: 300    # 5 minutes
8
    ReceiveMessageWaitTimeSeconds: 20 # Long polling
9
    KmsMasterKeyId: !Ref SQSKMSKey
10
    Tags:
11
      - Key: Environment
12
        Value: Production
13
      - Key: Application
14
        Value: MyApp
15

16
# Dead Letter Queue
17
DeadLetterQueue:
18
  Type: AWS::SQS::Queue
19
  Properties:
20
    QueueName: standard-queue-dlq
21
    MessageRetentionPeriod: 1209600
22

23
# Redrive Policy
24
RedrivePolicy:
25
  Type: AWS::SQS::QueuePolicy
26
  Properties:
27
    Queues:
28
      - !Ref StandardQueue
29
    PolicyDocument:
30
      Statement:
31
        - Effect: Allow
32
          Action: sqs:*
33
          Resource: !GetAtt StandardQueue.Arn
34
          Principal:
35
            AWS: !Sub "arn:aws:iam::${AWS::AccountId}:root"

1
# FIFO Queue for ordered processing
2
FIFOQueue:
3
  Type: AWS::SQS::Queue
4
  Properties:
5
    QueueName: fifo-queue.fifo
6
    FifoQueue: true
7
    ContentBasedDeduplication: true
8
    MessageRetentionPeriod: 1209600
9
    VisibilityTimeoutSeconds: 300
10
    KmsMasterKeyId: !Ref SQSKMSKey
11
    RedrivePolicy:
12
      deadLetterTargetArn: !GetAtt FIFODeadLetterQueue.Arn
13
      maxReceiveCount: 3
14

15
FIFODeadLetterQueue:
16
  Type: AWS::SQS::Queue
17
  Properties:
18
    QueueName: fifo-queue-dlq.fifo
19
    FifoQueue: true
20
    MessageRetentionPeriod: 1209600

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

6
sqs = boto3.client('sqs')
7

8
def send_standard_message(queue_url, message_body, attributes=None):
9
    """
10
    Send message to standard queue
11
    """
12
    params = {
13
        'QueueUrl': queue_url,
14
        'MessageBody': json.dumps(message_body) if isinstance(message_body, dict) else message_body
15
    }
16

17
    if attributes:
18
        params['MessageAttributes'] = {}
19
        for key, value in attributes.items():
20
            if isinstance(value, str):
21
                params['MessageAttributes'][key] = {
22
                    'StringValue': value,
23
                    'DataType': 'String'
24
                }
25
            elif isinstance(value, (int, float)):
26
                params['MessageAttributes'][key] = {
27
                    'StringValue': str(value),
28
                    'DataType': 'Number'
29
                }
30

31
    response = sqs.send_message(**params)
32
    return response['MessageId']
33

34
def send_fifo_message(queue_url, message_body, group_id, deduplication_id=None, attributes=None):
35
    """
36
    Send message to FIFO queue
37
    """
38
    params = {
39
        'QueueUrl': queue_url,
40
        'MessageBody': json.dumps(message_body) if isinstance(message_body, dict) else message_body,
41
        'MessageGroupId': group_id
42
    }
43

44
    if deduplication_id:
45
        params['MessageDeduplicationId'] = deduplication_id
46

47
    if attributes:
48
        params['MessageAttributes'] = {}
49
        for key, value in attributes.items():
50
            if isinstance(value, str):
51
                params['MessageAttributes'][key] = {
52
                    'StringValue': value,
53
                    'DataType': 'String'
54
                }
55

56
    response = sqs.send_message(**params)
57
    return response['MessageId']
58

59
def send_batch_messages(queue_url, messages, is_fifo=False):
60
    """
61
    Send multiple messages in batch (up to 10)
62
    """
63
    entries = []
64

65
    for i, message in enumerate(messages[:10]):  # Limit to 10 messages
66
        entry = {
67
            'Id': str(i),
68
            'MessageBody': json.dumps(message['body']) if isinstance(message['body'], dict) else message['body']
69
        }
70

71
        if message.get('attributes'):
72
            entry['MessageAttributes'] = {}
73
            for key, value in message['attributes'].items():
74
                entry['MessageAttributes'][key] = {
75
                    'StringValue': str(value),
76
                    'DataType': 'String' if isinstance(value, str) else 'Number'
77
                }
78

79
        if is_fifo:
80
            entry['MessageGroupId'] = message.get('group_id', 'default')
81
            if message.get('deduplication_id'):
82
                entry['MessageDeduplicationId'] = message['deduplication_id']
83

84
        entries.append(entry)
85

86
    response = sqs.send_message_batch(
87
        QueueUrl=queue_url,
88
        Entries=entries
89
    )
90

91
    return {
92
        'successful': response.get('Successful', []),
93
        'failed': response.get('Failed', [])
94
    }
95

96
# Example usage
97
queue_url = 'https://sqs.us-east-1.amazonaws.com/123456789012/standard-queue'
98

99
# Send standard message
100
message_id = send_standard_message(
101
    queue_url,
102
    {
103
        'order_id': 'ORDER-123456',
104
        'customer_id': 'CUST-789',
105
        'amount': 99.99,
106
        'timestamp': datetime.utcnow().isoformat()
107
    },
108
    attributes={
109
        'source': 'order-service',
110
        'priority': 'high'
111
    }
112
)
113

114
# Send FIFO message
115
fifo_queue_url = 'https://sqs.us-east-1.amazonaws.com/123456789012/fifo-queue.fifo'
116
message_id = send_fifo_message(
117
    fifo_queue_url,
118
    {'action': 'process_payment', 'order_id': 'ORDER-123456'},
119
    group_id='payment-processing',
120
    deduplication_id=f'payment-{uuid.uuid4()}'
121
)

1
import time
2
import json
3

4
def receive_messages(queue_url, max_messages=10, wait_time=20):
5
    """
6
    Receive messages from queue with long polling
7
    """
8
    response = sqs.receive_message(
9
        QueueUrl=queue_url,
10
        MaxNumberOfMessages=max_messages,
11
        WaitTimeSeconds=wait_time,
12
        MessageAttributeNames=['All'],
13
        AttributeNames=['All']
14
    )
15

16
    return response.get('Messages', [])
17

18
def process_messages(queue_url, processor_function):
19
    """
20
    Continuous message processing loop
21
    """
22
    while True:
23
        try:
24
            messages = receive_messages(queue_url)
25

26
            if not messages:
27
                print("No messages received, continuing...")
28
                continue
29

30
            for message in messages:
31
                try:
32
                    # Parse message body
33
                    body = json.loads(message['Body'])
34
                    receipt_handle = message['ReceiptHandle']
35
                    message_attributes = message.get('MessageAttributes', {})
36

37
                    # Process message
38
                    result = processor_function(body, message_attributes)
39

40
                    if result.get('success', False):
41
                        # Delete message after successful processing
42
                        delete_message(queue_url, receipt_handle)
43
                        print(f"Successfully processed message: {message['MessageId']}")
44
                    else:
45
                        print(f"Failed to process message: {message['MessageId']}")
46
                        # Message will become visible again after visibility timeout
47

48
                except json.JSONDecodeError:
49
                    print(f"Invalid JSON in message: {message['MessageId']}")
50
                    delete_message(queue_url, receipt_handle)  # Remove invalid message
51
                except Exception as e:
52
                    print(f"Error processing message {message['MessageId']}: {e}")
53

54
        except KeyboardInterrupt:
55
            print("Shutting down message processor...")
56
            break
57
        except Exception as e:
58
            print(f"Error receiving messages: {e}")
59
            time.sleep(5)  # Wait before retrying
60

61
def delete_message(queue_url, receipt_handle):
62
    """
63
    Delete processed message from queue
64
    """
65
    sqs.delete_message(
66
        QueueUrl=queue_url,
67
        ReceiptHandle=receipt_handle
68
    )
69

70
def delete_batch_messages(queue_url, messages):
71
    """
72
    Delete multiple messages in batch
73
    """
74
    entries = [
75
        {
76
            'Id': str(i),
77
            'ReceiptHandle': msg['ReceiptHandle']
78
        }
79
        for i, msg in enumerate(messages[:10])  # Limit to 10
80
    ]
81

82
    response = sqs.delete_message_batch(
83
        QueueUrl=queue_url,
84
        Entries=entries
85
    )
86

87
    return response
88

89
# Example message processor
90
def order_processor(message_body, attributes):
91
    """
92
    Example order processing function
93
    """
94
    try:
95
        order_id = message_body.get('order_id')
96
        customer_id = message_body.get('customer_id')
97
        amount = message_body.get('amount')
98

99
        print(f"Processing order {order_id} for customer {customer_id}, amount: ${amount}")
100

101
        # Simulate processing
102
        time.sleep(1)
103

104
        # Simulate random failures for testing
105
        import random
106
        if random.random() > 0.9:  # 10% failure rate
107
            raise Exception("Simulated processing error")
108

109
        return {'success': True}
110

111
    except Exception as e:
112
        print(f"Processing failed: {e}")
113
        return {'success': False}
114

115
# Start processing messages
116
# process_messages(queue_url, order_processor)

1
def change_message_visibility(queue_url, receipt_handle, visibility_timeout):
2
    """
3
    Change message visibility timeout
4
    """
5
    sqs.change_message_visibility(
6
        QueueUrl=queue_url,
7
        ReceiptHandle=receipt_handle,
8
        VisibilityTimeout=visibility_timeout
9
    )
10

11
def extend_processing_time(queue_url, receipt_handle, additional_seconds=300):
12
    """
13
    Extend message processing time
14
    """
15
    change_message_visibility(queue_url, receipt_handle, additional_seconds)
16

17
def process_with_extended_timeout(queue_url, processor_function):
18
    """
19
    Process messages with ability to extend timeout
20
    """
21
    messages = receive_messages(queue_url)
22

23
    for message in messages:
24
        try:
25
            body = json.loads(message['Body'])
26
            receipt_handle = message['ReceiptHandle']
27

28
            # Start processing
29
            start_time = time.time()
30

31
            # Check if processing is taking too long and extend timeout
32
            if time.time() - start_time > 240:  # If processing > 4 minutes
33
                extend_processing_time(queue_url, receipt_handle, 600)  # Extend by 10 minutes
34

35
            result = processor_function(body)
36

37
            if result.get('success'):
38
                delete_message(queue_url, receipt_handle)
39

40
        except Exception as e:
41
            print(f"Error processing message: {e}")

1
# Queue with dead letter queue configuration
2
MainQueue:
3
  Type: AWS::SQS::Queue
4
  Properties:
5
    QueueName: main-processing-queue
6
    RedrivePolicy:
7
      deadLetterTargetArn: !GetAtt MainDeadLetterQueue.Arn
8
      maxReceiveCount: 3  # Send to DLQ after 3 failed attempts
9
    VisibilityTimeoutSeconds: 300
10

11
MainDeadLetterQueue:
12
  Type: AWS::SQS::Queue
13
  Properties:
14
    QueueName: main-processing-dlq
15
    MessageRetentionPeriod: 1209600  # Keep for 14 days for analysis

1
def analyze_dead_letter_queue(dlq_url):
2
    """
3
    Analyze messages in dead letter queue
4
    """
5
    messages = receive_messages(dlq_url, max_messages=10)
6

7
    for message in messages:
8
        body = json.loads(message['Body'])
9
        attributes = message.get('Attributes', {})
10

11
        print(f"Dead Letter Message Analysis:")
12
        print(f"  Message ID: {message['MessageId']}")
13
        print(f"  First Received: {attributes.get('ApproximateFirstReceiveTimestamp')}")
14
        print(f"  Receive Count: {attributes.get('ApproximateReceiveCount')}")
15
        print(f"  Body: {json.dumps(body, indent=2)}")
16

17
        # Optionally reprocess or log for manual review
18
        # reprocess_failed_message(body)
19

20
def redrive_messages_from_dlq(source_dlq_url, target_queue_url, max_messages=10):
21
    """
22
    Move messages from DLQ back to main queue for reprocessing
23
    """
24
    messages = receive_messages(source_dlq_url, max_messages)
25
    redriven_count = 0
26

27
    for message in messages:
28
        try:
29
            # Send to target queue
30
            sqs.send_message(
31
                QueueUrl=target_queue_url,
32
                MessageBody=message['Body'],
33
                MessageAttributes=message.get('MessageAttributes', {})
34
            )
35

36
            # Delete from DLQ
37
            delete_message(source_dlq_url, message['ReceiptHandle'])
38
            redriven_count += 1
39

40
        except Exception as e:
41
            print(f"Failed to redrive message {message['MessageId']}: {e}")
42

43
    return redriven_count

1
def route_messages_by_type(queue_url, processors):
2
    """
3
    Route messages to different processors based on message type
4
    """
5
    messages = receive_messages(queue_url)
6

7
    for message in messages:
8
        try:
9
            body = json.loads(message['Body'])
10
            message_type = body.get('type', 'default')
11

12
            processor = processors.get(message_type, processors.get('default'))
13

14
            if processor:
15
                result = processor(body, message.get('MessageAttributes', {}))
16

17
                if result.get('success'):
18
                    delete_message(queue_url, message['ReceiptHandle'])
19
                else:
20
                    # Let message return to queue for retry
21
                    print(f"Processing failed for message type {message_type}")
22
            else:
23
                print(f"No processor found for message type: {message_type}")
24
                delete_message(queue_url, message['ReceiptHandle'])  # Remove unknown types
25

26
        except Exception as e:
27
            print(f"Error routing message: {e}")
28

29
# Example processors
30
processors = {
31
    'order': lambda body, attrs: {'success': process_order(body)},
32
    'payment': lambda body, attrs: {'success': process_payment(body)},
33
    'notification': lambda body, attrs: {'success': send_notification(body)},
34
    'default': lambda body, attrs: {'success': log_unknown_message(body)}
35
}
36

37
# route_messages_by_type(queue_url, processors)

1
def get_queue_metrics(queue_url):
2
    """
3
    Get queue metrics for monitoring
4
    """
5
    response = sqs.get_queue_attributes(
6
        QueueUrl=queue_url,
7
        AttributeNames=[
8
            'ApproximateNumberOfMessages',
9
            'ApproximateNumberOfMessagesNotVisible',
10
            'ApproximateNumberOfMessagesDelayed',
11
            'CreatedTimestamp',
12
            'LastModifiedTimestamp',
13
            'ApproximateAgeOfOldestMessage'
14
        ]
15
    )
16

17
    attributes = response['Attributes']
18

19
    return {
20
        'visible_messages': int(attributes.get('ApproximateNumberOfMessages', 0)),
21
        'in_flight_messages': int(attributes.get('ApproximateNumberOfMessagesNotVisible', 0)),
22
        'delayed_messages': int(attributes.get('ApproximateNumberOfMessagesDelayed', 0)),
23
        'oldest_message_age': int(attributes.get('ApproximateAgeOfOldestMessage', 0)),
24
        'total_messages': int(attributes.get('ApproximateNumberOfMessages', 0)) +
25
                         int(attributes.get('ApproximateNumberOfMessagesNotVisible', 0))
26
    }
27

28
def auto_scale_consumers(queue_url, min_consumers=1, max_consumers=10, target_messages_per_consumer=100):
29
    """
30
    Auto-scale consumers based on queue depth
31
    """
32
    metrics = get_queue_metrics(queue_url)
33
    total_messages = metrics['total_messages']
34

35
    # Calculate desired number of consumers
36
    desired_consumers = max(
37
        min_consumers,
38
        min(max_consumers, (total_messages // target_messages_per_consumer) + 1)
39
    )
40

41
    print(f"Queue metrics: {metrics}")
42
    print(f"Recommended consumers: {desired_consumers}")
43

44
    return desired_consumers
45

46
# Example CloudWatch custom metrics
47
def publish_queue_metrics(queue_name, metrics):
48
    """
49
    Publish custom metrics to CloudWatch
50
    """
51
    cloudwatch = boto3.client('cloudwatch')
52

53
    cloudwatch.put_metric_data(
54
        Namespace='SQS/Custom',
55
        MetricData=[
56
            {
57
                'MetricName': 'VisibleMessages',
58
                'Value': metrics['visible_messages'],
59
                'Unit': 'Count',
60
                'Dimensions': [
61
                    {'Name': 'QueueName', 'Value': queue_name}
62
                ]
63
            },
64
            {
65
                'MetricName': 'InFlightMessages',
66
                'Value': metrics['in_flight_messages'],
67
                'Unit': 'Count',
68
                'Dimensions': [
69
                    {'Name': 'QueueName', 'Value': queue_name}
70
                ]
71
            },
72
            {
73
                'MetricName': 'OldestMessageAge',
74
                'Value': metrics['oldest_message_age'],
75
                'Unit': 'Seconds',
76
                'Dimensions': [
77
                    {'Name': 'QueueName', 'Value': queue_name}
78
                ]
79
            }
80
        ]
81
    )

1
# Lambda function triggered by SQS
2
SQSProcessor:
3
  Type: AWS::Lambda::Function
4
  Properties:
5
    FunctionName: sqs-message-processor
6
    Runtime: python3.11
7
    Handler: index.lambda_handler
8
    Code:
9
      ZipFile: |
10
        import json
11
        import boto3
12

13
        def lambda_handler(event, context):
14
            processed = 0
15
            failed = 0
16

17
            for record in event['Records']:
18
                try:
19
                    body = json.loads(record['body'])
20
                    message_id = record['messageId']
21

22
                    # Process message
23
                    result = process_message(body)
24

25
                    if result['success']:
26
                        processed += 1
27
                    else:
28
                        failed += 1
29

30
                except Exception as e:
31
                    print(f"Error processing record: {e}")
32
                    failed += 1
33

34
            return {
35
                'statusCode': 200,
36
                'body': json.dumps({
37
                    'processed': processed,
38
                    'failed': failed
39
                })
40
            }
41

42
        def process_message(body):
43
            # Your processing logic here
44
            return {'success': True}
45

46
# Event source mapping
47
SQSEventSourceMapping:
48
  Type: AWS::Lambda::EventSourceMapping
49
  Properties:
50
    EventSourceArn: !GetAtt ProcessingQueue.Arn
51
    FunctionName: !GetAtt SQSProcessor.Arn
52
    BatchSize: 10
53
    MaximumBatchingWindowInSeconds: 5
54
    ReportBatchItemFailures: true

1
# SNS to SQS fanout pattern
2
NotificationTopic:
3
  Type: AWS::SNS::Topic
4
  Properties:
5
    TopicName: OrderEvents
6

7
# Multiple SQS queues for different processing
8
OrderProcessingQueue:
9
  Type: AWS::SQS::Queue
10
  Properties:
11
    QueueName: order-processing
12

13
EmailQueue:
14
  Type: AWS::SQS::Queue
15
  Properties:
16
    QueueName: email-notifications
17

18
AnalyticsQueue:
19
  Type: AWS::SQS::Queue
20
  Properties:
21
    QueueName: analytics-events
22

23
# SNS subscriptions
24
OrderProcessingSubscription:
25
  Type: AWS::SNS::Subscription
26
  Properties:
27
    Protocol: sqs
28
    Endpoint: !GetAtt OrderProcessingQueue.Arn
29
    TopicArn: !Ref NotificationTopic
30

31
EmailSubscription:
32
  Type: AWS::SNS::Subscription
33
  Properties:
34
    Protocol: sqs
35
    Endpoint: !GetAtt EmailQueue.Arn
36
    TopicArn: !Ref NotificationTopic

1
# Step Functions integration
2
def trigger_step_function_from_sqs():
3
    """
4
    Process SQS messages and trigger Step Functions
5
    """
6
    stepfunctions = boto3.client('stepfunctions')
7

8
    messages = receive_messages(queue_url)
9

10
    for message in messages:
11
        try:
12
            body = json.loads(message['Body'])
13

14
            # Start Step Function execution
15
            response = stepfunctions.start_execution(
16
                stateMachineArn='arn:aws:states:region:account:stateMachine:MyWorkflow',
17
                name=f"execution-{message['MessageId']}",
18
                input=json.dumps({
19
                    'messageId': message['MessageId'],
20
                    'payload': body,
21
                    'source': 'sqs'
22
                })
23
            )
24

25
            print(f"Started execution: {response['executionArn']}")
26
            delete_message(queue_url, message['ReceiptHandle'])
27

28
        except Exception as e:
29
            print(f"Failed to process message: {e}")

1
class BatchMessageProcessor:
2
    def __init__(self, queue_url, batch_size=10):
3
        self.queue_url = queue_url
4
        self.batch_size = batch_size
5
        self.sqs = boto3.client('sqs')
6

7
    def process_in_batches(self, processor_function):
8
        """
9
        Process messages in batches for better throughput
10
        """
11
        while True:
12
            # Receive batch of messages
13
            messages = self.receive_message_batch()
14

15
            if not messages:
16
                time.sleep(1)
17
                continue
18

19
            # Process batch
20
            results = self.process_batch(messages, processor_function)
21

22
            # Delete successfully processed messages
23
            self.cleanup_batch(messages, results)
24

25
    def receive_message_batch(self):
26
        """
27
        Receive up to batch_size messages
28
        """
29
        response = self.sqs.receive_message(
30
            QueueUrl=self.queue_url,
31
            MaxNumberOfMessages=self.batch_size,
32
            WaitTimeSeconds=20,
33
            MessageAttributeNames=['All']
34
        )
35
        return response.get('Messages', [])
36

37
    def process_batch(self, messages, processor_function):
38
        """
39
        Process all messages in batch
40
        """
41
        results = []
42

43
        for message in messages:
44
            try:
45
                body = json.loads(message['Body'])
46
                result = processor_function(body)
47
                results.append({
48
                    'message': message,
49
                    'success': result.get('success', False)
50
                })
51
            except Exception as e:
52
                results.append({
53
                    'message': message,
54
                    'success': False,
55
                    'error': str(e)
56
                })
57

58
        return results
59

60
    def cleanup_batch(self, messages, results):
61
        """
62
        Delete successfully processed messages in batch
63
        """
64
        successful_messages = [
65
            r['message'] for r in results if r['success']
66
        ]
67

68
        if successful_messages:
69
            delete_batch_messages(self.queue_url, successful_messages)
70

71
# Usage
72
processor = BatchMessageProcessor(queue_url, batch_size=10)
73
processor.process_in_batches(order_processor)

1
import threading
2
from concurrent.futures import ThreadPoolExecutor
3
from queue import Queue
4

5
class SQSConnectionPool:
6
    def __init__(self, max_connections=10):
7
        self.connections = Queue(maxsize=max_connections)
8
        self.max_connections = max_connections
9
        self._initialize_connections()
10

11
    def _initialize_connections(self):
12
        """
13
        Initialize connection pool
14
        """
15
        for _ in range(self.max_connections):
16
            conn = boto3.client('sqs')
17
            self.connections.put(conn)
18

19
    def get_connection(self):
20
        """
21
        Get connection from pool
22
        """
23
        return self.connections.get()
24

25
    def return_connection(self, conn):
26
        """
27
        Return connection to pool
28
        """
29
        self.connections.put(conn)
30

31
    def process_with_pool(self, queue_url, processor_function, num_workers=5):
32
        """
33
        Process messages using connection pool and multiple workers
34
        """
35
        with ThreadPoolExecutor(max_workers=num_workers) as executor:
36
            futures = []
37

38
            for _ in range(num_workers):
39
                future = executor.submit(
40
                    self._worker_process,
41
                    queue_url,
42
                    processor_function
43
                )
44
                futures.append(future)
45

46
            # Wait for all workers to complete
47
            for future in futures:
48
                try:
49
                    future.result()
50
                except Exception as e:
51
                    print(f"Worker failed: {e}")
52

53
    def _worker_process(self, queue_url, processor_function):
54
        """
55
        Individual worker processing messages
56
        """
57
        while True:
58
            conn = self.get_connection()
59
            try:
60
                response = conn.receive_message(
61
                    QueueUrl=queue_url,
62
                    MaxNumberOfMessages=10,
63
                    WaitTimeSeconds=20
64
                )
65

66
                messages = response.get('Messages', [])
67

68
                for message in messages:
69
                    try:
70
                        body = json.loads(message['Body'])
71
                        result = processor_function(body)
72

73
                        if result.get('success'):
74
                            conn.delete_message(
75
                                QueueUrl=queue_url,
76
                                ReceiptHandle=message['ReceiptHandle']
77
                            )
78
                    except Exception as e:
79
                        print(f"Processing error: {e}")
80

81
            finally:
82
                self.return_connection(conn)
83

84
# Usage
85
pool = SQSConnectionPool(max_connections=10)
86
pool.process_with_pool(queue_url, order_processor, num_workers=5)

1
# CloudWatch alarms for SQS monitoring
2
QueueDepthAlarm:
3
  Type: AWS::CloudWatch::Alarm
4
  Properties:
5
    AlarmName: SQS-High-Queue-Depth
6
    AlarmDescription: SQS queue has too many messages
7
    MetricName: ApproximateNumberOfVisibleMessages
8
    Namespace: AWS/SQS
9
    Statistic: Average
10
    Period: 300
11
    EvaluationPeriods: 2
12
    Threshold: 1000
13
    ComparisonOperator: GreaterThanThreshold
14
    Dimensions:
15
      - Name: QueueName
16
        Value: !GetAtt ProcessingQueue.QueueName
17

18
OldMessagesAlarm:
19
  Type: AWS::CloudWatch::Alarm
20
  Properties:
21
    AlarmName: SQS-Old-Messages
22
    AlarmDescription: Messages are too old in queue
23
    MetricName: ApproximateAgeOfOldestMessage
24
    Namespace: AWS/SQS
25
    Statistic: Maximum
26
    Period: 300
27
    EvaluationPeriods: 1
28
    Threshold: 3600  # 1 hour
29
    ComparisonOperator: GreaterThanThreshold
30
    Dimensions:
31
      - Name: QueueName
32
        Value: !GetAtt ProcessingQueue.QueueName

1
def analyze_queue_performance(queue_url, duration_minutes=60):
2
    """
3
    Analyze queue performance over time
4
    """
5
    cloudwatch = boto3.client('cloudwatch')
6
    end_time = datetime.utcnow()
7
    start_time = end_time - timedelta(minutes=duration_minutes)
8

9
    metrics = [
10
        'ApproximateNumberOfVisibleMessages',
11
        'ApproximateNumberOfMessagesNotVisible',
12
        'NumberOfMessagesSent',
13
        'NumberOfMessagesReceived',
14
        'NumberOfMessagesDeleted'
15
    ]
16

17
    queue_name = queue_url.split('/')[-1]
18
    results = {}
19

20
    for metric in metrics:
21
        response = cloudwatch.get_metric_statistics(
22
            Namespace='AWS/SQS',
23
            MetricName=metric,
24
            Dimensions=[
25
                {'Name': 'QueueName', 'Value': queue_name}
26
            ],
27
            StartTime=start_time,
28
            EndTime=end_time,
29
            Period=300,
30
            Statistics=['Average', 'Maximum', 'Sum']
31
        )
32

33
        results[metric] = {
34
            'datapoints': response['Datapoints'],
35
            'average': sum(p['Average'] for p in response['Datapoints']) / len(response['Datapoints']) if response['Datapoints'] else 0,
36
            'maximum': max(p['Maximum'] for p in response['Datapoints']) if response['Datapoints'] else 0
37
        }
38

39
    return results
40

41
def diagnose_queue_issues(queue_url):
42
    """
43
    Diagnose common queue issues
44
    """
45
    issues = []
46
    metrics = get_queue_metrics(queue_url)
47

48
    # Check for message buildup
49
    if metrics['visible_messages'] > 10000:
50
        issues.append("High number of visible messages - consider scaling consumers")
51

52
    # Check for old messages
53
    if metrics['oldest_message_age'] > 3600:  # 1 hour
54
        issues.append("Old messages detected - check consumer performance")
55

56
    # Check for stuck messages
57
    if metrics['in_flight_messages'] > metrics['visible_messages']:
58
        issues.append("High number of in-flight messages - check visibility timeout")
59

60
    # Check queue attributes for configuration issues
61
    response = sqs.get_queue_attributes(
62
        QueueUrl=queue_url,
63
        AttributeNames=['All']
64
    )
65

66
    attributes = response['Attributes']
67

68
    if int(attributes.get('VisibilityTimeoutSeconds', 0)) < 30:
69
        issues.append("Visibility timeout may be too short")
70

71
    if int(attributes.get('ReceiveMessageWaitTimeSeconds', 0)) == 0:
72
        issues.append("Long polling not enabled - consider enabling for cost savings")
73

74
    return {
75
        'metrics': metrics,
76
        'issues': issues,
77
        'recommendations': generate_recommendations(metrics, issues)
78
    }
79

80
def generate_recommendations(metrics, issues):
81
    """
82
    Generate performance recommendations
83
    """
84
    recommendations = []
85

86
    if metrics['visible_messages'] > 5000:
87
        recommendations.append("Scale up consumers or increase batch processing")
88

89
    if metrics['oldest_message_age'] > 1800:  # 30 minutes
90
        recommendations.append("Optimize message processing time or increase parallelism")
91

92
    if len(issues) == 0:
93
        recommendations.append("Queue performance looks healthy")
94

95
    return recommendations

1
# Queue access policy
2
QueueAccessPolicy:
3
  Type: AWS::SQS::QueuePolicy
4
  Properties:
5
    Queues:
6
      - !Ref SecureQueue
7
    PolicyDocument:
8
      Statement:
9
        - Effect: Allow
10
          Principal:
11
            AWS: !GetAtt ProcessingRole.Arn
12
          Action:
13
            - sqs:ReceiveMessage
14
            - sqs:DeleteMessage
15
            - sqs:ChangeMessageVisibility
16
          Resource: !GetAtt SecureQueue.Arn
17
        - Effect: Allow
18
          Principal:
19
            AWS: !GetAtt PublisherRole.Arn
20
          Action:
21
            - sqs:SendMessage
22
          Resource: !GetAtt SecureQueue.Arn
23
          Condition:
24
            StringEquals:
25
              'aws:SourceAccount': !Ref 'AWS::AccountId'

1
# KMS encrypted queue
2
EncryptedQueue:
3
  Type: AWS::SQS::Queue
4
  Properties:
5
    QueueName: encrypted-queue
6
    KmsMasterKeyId: !Ref SQSKMSKey
7
    KmsDataKeyReusePeriodSeconds: 300
8

9
SQSKMSKey:
10
  Type: AWS::KMS::Key
11
  Properties:
12
    Description: KMS key for SQS encryption
13
    KeyPolicy:
14
      Statement:
15
        - Effect: Allow
16
          Principal:
17
            AWS: !Sub "arn:aws:iam::${AWS::AccountId}:root"
18
          Action: "kms:*"
19
          Resource: "*"

1
def optimize_queue_costs(queue_url):
2
    """
3
    Analyze and optimize queue costs
4
    """
5
    # Analyze message patterns
6
    metrics = analyze_queue_performance(queue_url, duration_minutes=1440)  # 24 hours
7

8
    recommendations = []
9

10
    # Check polling efficiency
11
    sent = metrics['NumberOfMessagesSent']['average']
12
    received = metrics['NumberOfMessagesReceived']['average']
13

14
    if received > sent * 2:  # High receive-to-send ratio
15
        recommendations.append("Enable long polling to reduce API calls")
16

17
    # Check batch processing opportunities
18
    visible_messages = metrics['ApproximateNumberOfVisibleMessages']['average']
19

20
    if visible_messages > 100:
21
        recommendations.append("Consider batch processing to improve efficiency")
22

23
    # Estimate monthly costs
24
    monthly_requests = (sent + received) * 30 * 24 * 60 / 5  # Assuming 5-minute intervals
25
    monthly_cost = monthly_requests * 0.0000004  # $0.40 per million requests
26

27
    return {
28
        'current_monthly_cost_estimate': monthly_cost,
29
        'recommendations': recommendations,
30
        'metrics_summary': {
31
            'avg_sent_per_hour': sent,
32
            'avg_received_per_hour': received,
33
            'avg_visible_messages': visible_messages
34
        }
35
    }
36

37
def implement_long_polling(queue_url):
38
    """
39
    Enable long polling to reduce costs
40
    """
41
    sqs.set_queue_attributes(
42
        QueueUrl=queue_url,
43
        Attributes={
44
            'ReceiveMessageWaitTimeSeconds': '20'  # Maximum long polling duration
45
        }
46
    )
47

48
    print(f"Enabled long polling for queue: {queue_url}")