fix(plugins): Add MessageGroup to SQS (#502)

* fix(plugins): Add MessageGroup to SQS

Author: Stephen Cefali

src/sentry_plugins/amazon_sqs/plugin.py

@@ -4,7 +4,6 @@
 
 import boto3
 from botocore.client import ClientError
-
 from sentry_plugins.base import CorePluginMixin
 from sentry.plugins.bases.data_forwarding import DataForwardingPlugin
 from sentry_plugins.utils import get_secret_field_config
@@ -43,13 +42,21 @@ def get_config(self, project, **kwargs):
             get_secret_field_config(
                 name="secret_key", label="Secret Key", secret=self.get_option("secret_key", project)
             ),
+            {
+                "name": "message_group_id",
+                "label": "Message Group ID",
+                "type": "text",
+                "required": False,
+                "placeholder": "Required for FIFO queues, exclude for standard queues",
+            },
         ]
 
     def forward_event(self, event, payload):
         queue_url = self.get_option("queue_url", event.project)
         access_key = self.get_option("access_key", event.project)
         secret_key = self.get_option("secret_key", event.project)
         region = self.get_option("region", event.project)
+        message_group_id = self.get_option("message_group_id", event.project)
 
         if not all((queue_url, access_key, secret_key, region)):
             return
@@ -68,7 +75,20 @@ def forward_event(self, event, payload):
                 aws_secret_access_key=secret_key,
                 region_name=region,
             )
-            client.send_message(QueueUrl=queue_url, MessageBody=message)
+
+            message = {"QueueUrl": queue_url, "MessageBody": message}
+
+            # need a MessageGroupId for FIFO queues
+            # note that if MessageGroupId is specified for non-FIFO, this will fail
+            if message_group_id:
+                from uuid import uuid4
+
+                message["MessageGroupId"] = message_group_id
+                # if content based de-duplication is not enabled, we need to provide a
+                # MessageDeduplicationId
+                message["MessageDeduplicationId"] = uuid4().hex
+
+            client.send_message(**message)
         except ClientError as e:
             if e.message.startswith("An error occurred (AccessDenied)"):
                 # If there's an issue with the user's token then we can't do
@@ -92,6 +112,27 @@ def forward_event(self, event, payload):
                     },
                 )
                 return False
+            elif e.message.endswith("must contain the parameter MessageGroupId."):
+                metrics_name = "sentry_plugins.amazon_sqs.missing_message_group_id"
+                logger.info(
+                    metrics_name,
+                    extra={
+                        "queue_url": queue_url,
+                        "access_key": access_key,
+                        "region": region,
+                        "project_id": event.project.id,
+                        "organization_id": event.project.organization_id,
+                        "message_group_id": message_group_id,
+                    },
+                )
+                metrics.incr(
+                    metrics_name,
+                    tags={
+                        "project_id": event.project_id,
+                        "organization_id": event.project.organization_id,
+                    },
+                )
+                return False
             raise
 
         return True

tests/amazon_sqs/test_plugin.py

@@ -77,3 +77,41 @@ def test_token_error(self, mock_client, logger):
         assert (
             logger.info.call_args_list[0][0][0] == "sentry_plugins.amazon_sqs.access_token_invalid"
         )
+
+    @patch("sentry_plugins.amazon_sqs.plugin.logger")
+    @patch("boto3.client")
+    def test_message_group_error(self, mock_client, logger):
+        mock_client.return_value.send_message.side_effect = ClientError(
+            {
+                "Error": {
+                    "Code": "MissingParameter",
+                    "Message": "The request must contain the parameter MessageGroupId.",
+                }
+            },
+            "SendMessage",
+        )
+
+        self.run_test()
+
+        assert len(logger.info.call_args_list) == 1
+        assert (
+            logger.info.call_args_list[0][0][0]
+            == "sentry_plugins.amazon_sqs.missing_message_group_id"
+        )
+
+    @patch("uuid.uuid4")
+    @patch("boto3.client")
+    def test_pass_message_group_id(self, mock_client, mock_uuid):
+        class uuid(object):
+            hex = "some-uuid"
+
+        mock_uuid.return_value = uuid
+        self.plugin.set_option("message_group_id", "my_group", self.project)
+        event = self.run_test()
+
+        mock_client.return_value.send_message.assert_called_once_with(
+            QueueUrl="https://sqs-us-east-1.amazonaws.com/12345678/myqueue",
+            MessageBody=json.dumps(self.plugin.get_event_payload(event)),
+            MessageGroupId="my_group",
+            MessageDeduplicationId="some-uuid",
+        )