Repository: cloud-custodian/cloud-custodian

Issue:
Add a way to filter out dead-letter-queues from list of SQS queues
### Describe the feature

We are looking [for](url) a way to cleanup our unused AWS SQS queues. We have a policy that filters down to queues that have not been used in the past year. Now we need to be able to filter out dead-letter-queues (DLQs) from the list. We do not want to inadvertently remove any DLQs that are still needed but not receiving messages (because everything works so well).

We had a quick discussion about this in Gitter on May 26-27 and @ajkerrigan gave us a way to identify the DLQs using jq. Hopefully something can be added to custodian to allow us to filter out the DLQs directly from the policy?

### Extra information or context

Here is a bit of what @ajkerrigan had to say in Gitter:

>  this sounds like a feature request for a filter that checks for dead letter source queues. It could use list-dead-letter-source-queues under the hood (alternatively since we already have redrive policies cached for all queues in the region, may be able to leverage that data and save on API calls). It's worth filing as an issue to sort it out though. Ananth's workaround looking for the queue twice in resources.json is interesting though in the meantime.

Discussion:
thisisshi: Ugh... placebo and MacOS not playing nicely, reading the debug logs it seems that sometimes placebo does not actually read all of the flight recordings:

```
DEBUG    placebo.pill:pill.py:141 attaching to session: Session(region_name='us-east-1')
DEBUG    placebo.pill:pill.py:142 datapath: /Users/sonny/dev/thisisshi/cloud-custodian/tests/data/placebo/test_sqs_deadletter_filter
DEBUG    placebo.pill:pill.py:134 _create_client
DEBUG    placebo.pill:pill.py:300 _make_request: sqs.ListQueues
DEBUG    placebo.pill:pill.py:283 load_response: sqs.ListQueues
DEBUG    placebo.pill:pill.py:243 get_next_file_path: sqs.ListQueues
DEBUG    placebo.pill:pill.py:286 load_response: /Users/sonny/dev/thisisshi/cloud-custodian/tests/data/placebo/test_sqs_deadletter_filter/sqs.ListQueues_1.json
DEBUG    placebo.pill:pill.py:134 _create_client
DEBUG    placebo.pill:pill.py:300 _make_request: sqs.GetQueueAttributes
DEBUG    placebo.pill:pill.py:283 load_response: sqs.GetQueueAttributes
DEBUG    placebo.pill:pill.py:243 get_next_file_path: sqs.GetQueueAttributes
DEBUG    placebo.pill:pill.py:300 _make_request: sqs.GetQueueAttributes
DEBUG    placebo.pill:pill.py:283 load_response: sqs.GetQueueAttributes
DEBUG    placebo.pill:pill.py:243 get_next_file_path: sqs.GetQueueAttributes
DEBUG    placebo.pill:pill.py:286 load_response: /Users/sonny/dev/thisisshi/cloud-custodian/tests/data/placebo/test_sqs_deadletter_filter/sqs.GetQueueAttributes_1.json
DEBUG    placebo.pill:pill.py:286 load_response: /Users/sonny/dev/thisisshi/cloud-custodian/tests/data/placebo/test_sqs_deadletter_filter/sqs.GetQueueAttributes_1.json
DEBUG    placebo.pill:pill.py:300 _make_request: sqs.GetQueueAttributes
DEBUG    placebo.pill:pill.py:283 load_response: sqs.GetQueueAttributes
DEBUG    placebo.pill:pill.py:243 get_next_file_path: sqs.GetQueueAttributes
DEBUG    placebo.pill:pill.py:286 load_response: /Users/sonny/dev/thisisshi/cloud-custodian/tests/data/placebo/test_sqs_deadletter_filter/sqs.GetQueueAttributes_3.json
DEBUG    placebo.pill:pill.py:300 _make_request: sqs.GetQueueAttributes
DEBUG    placebo.pill:pill.py:283 load_response: sqs.GetQueueAttributes
DEBUG    placebo.pill:pill.py:243 get_next_file_path: sqs.GetQueueAttributes
DEBUG    placebo.pill:pill.py:286 load_response: /Users/sonny/dev/thisisshi/cloud-custodian/tests/data/placebo/test_sqs_deadletter_filter/sqs.GetQueueAttributes_4.json
```

note how `sqs.GetQueueAttributes_1.json` is loaded twice here.
thisisshi: later on in the tail end of the log when we pull the resources again, we see that the 4th response isn't loaded:

```
DEBUG    placebo.pill:pill.py:286 load_response: /Users/sonny/dev/thisisshi/cloud-custodian/tests/data/placebo/test_sqs_deadletter_filter/sqs.GetQueueAttributes_1.json
DEBUG    placebo.pill:pill.py:243 get_next_file_path: sqs.GetQueueAttributes
DEBUG    placebo.pill:pill.py:286 load_response: /Users/sonny/dev/thisisshi/cloud-custodian/tests/data/placebo/test_sqs_deadletter_filter/sqs.GetQueueAttributes_2.json
DEBUG    placebo.pill:pill.py:300 _make_request: sqs.GetQueueAttributes
DEBUG    placebo.pill:pill.py:283 load_response: sqs.GetQueueAttributes
DEBUG    placebo.pill:pill.py:243 get_next_file_path: sqs.GetQueueAttributes
DEBUG    placebo.pill:pill.py:300 _make_request: sqs.GetQueueAttributes
DEBUG    placebo.pill:pill.py:283 load_response: sqs.GetQueueAttributes
DEBUG    placebo.pill:pill.py:243 get_next_file_path: sqs.GetQueueAttributes
DEBUG    placebo.pill:pill.py:286 load_response: /Users/sonny/dev/thisisshi/cloud-custodian/tests/data/placebo/test_sqs_deadletter_filter/sqs.GetQueueAttributes_3.json
DEBUG    placebo.pill:pill.py:286 load_response: /Users/sonny/dev/thisisshi/cloud-custodian/tests/data/placebo/test_sqs_deadletter_filter/sqs.GetQueueAttributes_3.json
```
kapilt: the read twice might be a concurrency issue

Patch:
diff --git a/c7n/resources/sqs.py b/c7n/resources/sqs.py
index 606d1e090..c910d51cc 100644
--- a/c7n/resources/sqs.py
+++ b/c7n/resources/sqs.py
@@ -6,6 +6,7 @@ import json
 
 from c7n.actions import RemovePolicyBase, ModifyPolicyBase
 from c7n.filters import CrossAccountAccessFilter, MetricsFilter
+from c7n.filters.core import Filter
 from c7n.filters.kms import KmsRelatedFilter
 from c7n.manager import resources
 from c7n.utils import local_session
@@ -371,3 +372,47 @@ class SetRetentionPeriod(BaseAction):
                 QueueUrl=q['QueueUrl'],
                 Attributes={
                     'MessageRetentionPeriod': period})
+
+
+@SQS.filter_registry.register('dead-letter')
+class DeadLetterFilter(Filter):
+    """
+    Filter for sqs queues that are dead letter queues
+
+    :example:
+
+    .. code-block:: yaml
+
+        policies:
+         - name: find-dead-letter-queues
+           resource: aws.sqs
+           filters:
+             - type: dead-letter
+    """
+
+    schema = type_schema('dead-letter')
+    permissions = ()
+
+    def process(self, resources, event=None):
+        # we need to inspect all the queues regardless of any filters that
+        # may have been applied earlier
+        all_resources = self.manager.get_resource_manager("sqs").resources()
+        all_queue_arn_map = {r['QueueArn']: r for r in all_resources}
+        queue_arn_map = {r['QueueArn']: r for r in resources}
+        has_redrive = []
+        for r in all_resources:
+            if r.get("RedrivePolicy"):
+                has_redrive.append(r['QueueArn'])
+        result = []
+        # dead letter queues must exist in the same region and account as the
+        # original queue so it should be safe to look for them in our existing
+        # resources
+        for r in all_resources:
+            if r['QueueArn'] in has_redrive:
+                queue = all_queue_arn_map[r['QueueArn']]
+                target = json.loads(queue['RedrivePolicy']).get('deadLetterTargetArn')
+                if queue_arn_map.get(target):
+                    result.append(target)
+        # in case there are multiple queues pointing at the same dead letter queue
+        # we need to only return the unique queues
+        return [queue_arn_map[r] for r in set(result)]


Test patch:
diff --git a/tests/data/placebo/test_sqs_deadletter_filter/sqs.GetQueueAttributes_1.json b/tests/data/placebo/test_sqs_deadletter_filter/sqs.GetQueueAttributes_1.json
new file mode 100644
index 000000000..351020308
--- /dev/null
+++ b/tests/data/placebo/test_sqs_deadletter_filter/sqs.GetQueueAttributes_1.json
@@ -0,0 +1,24 @@
+{
+    "status_code": 200,
+    "data": {
+        "Attributes": {
+            "QueueArn": "arn:aws:sqs:us-east-1:644160558196:bar",
+            "ApproximateNumberOfMessages": "0",
+            "ApproximateNumberOfMessagesNotVisible": "0",
+            "ApproximateNumberOfMessagesDelayed": "0",
+            "CreatedTimestamp": "1648601669",
+            "LastModifiedTimestamp": "1648601750",
+            "VisibilityTimeout": "450",
+            "MaximumMessageSize": "262144",
+            "MessageRetentionPeriod": "345600",
+            "DelaySeconds": "0",
+            "Policy": "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Sid\":\"\",\"Effect\":\"Allow\",\"Principal\":{\"Service\":\"sns.amazonaws.com\"},\"Action\":\"sqs:SendMessage\",\"Resource\":\"arn:aws:sqs:us-east-1:644160558196:bar\",\"Condition\":{\"ArnEquals\":{\"aws:SourceArn\":\"arn:aws:sns:*:644160558196:*\"}}}]}",
+            "RedrivePolicy": "{\"deadLetterTargetArn\":\"arn:aws:sqs:us-east-1:644160558196:bar_dlq\",\"maxReceiveCount\":3}",
+            "ReceiveMessageWaitTimeSeconds": "0",
+            "KmsMasterKeyId": "arn:aws:kms:us-east-1:644160558196:key/75a0b693-ba69-4934-882b-10ba0f457567",
+            "KmsDataKeyReusePeriodSeconds": "3600",
+            "SqsManagedSseEnabled": "false"
+        },
+        "ResponseMetadata": {}
+    }
+}
diff --git a/tests/data/placebo/test_sqs_deadletter_filter/sqs.GetQueueAttributes_2.json b/tests/data/placebo/test_sqs_deadletter_filter/sqs.GetQueueAttributes_2.json
new file mode 100644
index 000000000..dcd295f57
--- /dev/null
+++ b/tests/data/placebo/test_sqs_deadletter_filter/sqs.GetQueueAttributes_2.json
@@ -0,0 +1,22 @@
+{
+    "status_code": 200,
+    "data": {
+        "Attributes": {
+            "QueueArn": "arn:aws:sqs:us-east-1:644160558196:bar_dlq",
+            "ApproximateNumberOfMessages": "0",
+            "ApproximateNumberOfMessagesNotVisible": "0",
+            "ApproximateNumberOfMessagesDelayed": "0",
+            "CreatedTimestamp": "1648601612",
+            "LastModifiedTimestamp": "1648601612",
+            "VisibilityTimeout": "150",
+            "MaximumMessageSize": "262144",
+            "MessageRetentionPeriod": "345600",
+            "DelaySeconds": "0",
+            "ReceiveMessageWaitTimeSeconds": "0",
+            "KmsMasterKeyId": "arn:aws:kms:us-east-1:644160558196:key/75a0b693-ba69-4934-882b-10ba0f457567",
+            "KmsDataKeyReusePeriodSeconds": "3600",
+            "SqsManagedSseEnabled": "false"
+        },
+        "ResponseMetadata": {}
+    }
+}
diff --git a/tests/data/placebo/test_sqs_deadletter_filter/sqs.GetQueueAttributes_3.json b/tests/data/placebo/test_sqs_deadletter_filter/sqs.GetQueueAttributes_3.json
new file mode 100644
index 000000000..33313a623
--- /dev/null
+++ b/tests/data/placebo/test_sqs_deadletter_filter/sqs.GetQueueAttributes_3.json
@@ -0,0 +1,22 @@
+{
+    "status_code": 200,
+    "data": {
+        "Attributes": {
+            "QueueArn": "arn:aws:sqs:us-east-1:644160558196:test",
+            "ApproximateNumberOfMessages": "0",
+            "ApproximateNumberOfMessagesNotVisible": "0",
+            "ApproximateNumberOfMessagesDelayed": "0",
+            "CreatedTimestamp": "1654623869",
+            "LastModifiedTimestamp": "1654623869",
+            "VisibilityTimeout": "30",
+            "MaximumMessageSize": "262144",
+            "MessageRetentionPeriod": "345600",
+            "DelaySeconds": "0",
+            "Policy": "{\"Version\":\"2008-10-17\",\"Id\":\"__default_policy_ID\",\"Statement\":[{\"Sid\":\"__owner_statement\",\"Effect\":\"Allow\",\"Principal\":{\"AWS\":\"arn:aws:iam::644160558196:root\"},\"Action\":\"SQS:*\",\"Resource\":\"arn:aws:sqs:us-east-1:644160558196:test\"}]}",
+            "RedrivePolicy": "{\"deadLetterTargetArn\":\"arn:aws:sqs:us-east-1:644160558196:test-dlq\",\"maxReceiveCount\":10}",
+            "ReceiveMessageWaitTimeSeconds": "0",
+            "SqsManagedSseEnabled": "false"
+        },
+        "ResponseMetadata": {}
+    }
+}
diff --git a/tests/data/placebo/test_sqs_deadletter_filter/sqs.GetQueueAttributes_4.json b/tests/data/placebo/test_sqs_deadletter_filter/sqs.GetQueueAttributes_4.json
new file mode 100644
index 000000000..625d28355
--- /dev/null
+++ b/tests/data/placebo/test_sqs_deadletter_filter/sqs.GetQueueAttributes_4.json
@@ -0,0 +1,21 @@
+{
+    "status_code": 200,
+    "data": {
+        "Attributes": {
+            "QueueArn": "arn:aws:sqs:us-east-1:644160558196:test-dlq",
+            "ApproximateNumberOfMessages": "0",
+            "ApproximateNumberOfMessagesNotVisible": "0",
+            "ApproximateNumberOfMessagesDelayed": "0",
+            "CreatedTimestamp": "1654623851",
+            "LastModifiedTimestamp": "1654623851",
+            "VisibilityTimeout": "30",
+            "MaximumMessageSize": "262144",
+            "MessageRetentionPeriod": "345600",
+            "DelaySeconds": "0",
+            "Policy": "{\"Version\":\"2008-10-17\",\"Id\":\"__default_policy_ID\",\"Statement\":[{\"Sid\":\"__owner_statement\",\"Effect\":\"Allow\",\"Principal\":{\"AWS\":\"arn:aws:iam::644160558196:root\"},\"Action\":\"SQS:*\",\"Resource\":\"arn:aws:sqs:us-east-1:644160558196:test-dlq\"}]}",
+            "ReceiveMessageWaitTimeSeconds": "0",
+            "SqsManagedSseEnabled": "false"
+        },
+        "ResponseMetadata": {}
+    }
+}
diff --git a/tests/data/placebo/test_sqs_deadletter_filter/sqs.ListQueues_1.json b/tests/data/placebo/test_sqs_deadletter_filter/sqs.ListQueues_1.json
new file mode 100644
index 000000000..7d0cc410b
--- /dev/null
+++ b/tests/data/placebo/test_sqs_deadletter_filter/sqs.ListQueues_1.json
@@ -0,0 +1,12 @@
+{
+    "status_code": 200,
+    "data": {
+        "QueueUrls": [
+            "https://us-east-1.queue.amazonaws.com/644160558196/bar",
+            "https://us-east-1.queue.amazonaws.com/644160558196/bar_dlq",
+            "https://us-east-1.queue.amazonaws.com/644160558196/test",
+            "https://us-east-1.queue.amazonaws.com/644160558196/test-dlq"
+        ],
+        "ResponseMetadata": {}
+    }
+}
diff --git a/tests/data/placebo/test_sqs_deadletter_filter/sqs.ListQueues_2.json b/tests/data/placebo/test_sqs_deadletter_filter/sqs.ListQueues_2.json
new file mode 100644
index 000000000..7d0cc410b
--- /dev/null
+++ b/tests/data/placebo/test_sqs_deadletter_filter/sqs.ListQueues_2.json
@@ -0,0 +1,12 @@
+{
+    "status_code": 200,
+    "data": {
+        "QueueUrls": [
+            "https://us-east-1.queue.amazonaws.com/644160558196/bar",
+            "https://us-east-1.queue.amazonaws.com/644160558196/bar_dlq",
+            "https://us-east-1.queue.amazonaws.com/644160558196/test",
+            "https://us-east-1.queue.amazonaws.com/644160558196/test-dlq"
+        ],
+        "ResponseMetadata": {}
+    }
+}
diff --git a/tests/data/placebo/test_sqs_deadletter_filter/tagging.GetResources_1.json b/tests/data/placebo/test_sqs_deadletter_filter/tagging.GetResources_1.json
new file mode 100644
index 000000000..2a346fb6f
--- /dev/null
+++ b/tests/data/placebo/test_sqs_deadletter_filter/tagging.GetResources_1.json
@@ -0,0 +1,40 @@
+{
+    "status_code": 200,
+    "data": {
+        "PaginationToken": "",
+        "ResourceTagMappingList": [
+            {
+                "ResourceARN": "arn:aws:sqs:us-east-1:644160558196:bar",
+                "Tags": [
+                    {
+                        "Key": "App",
+                        "Value": "dev"
+                    }
+                ]
+            },
+            {
+                "ResourceARN": "arn:aws:sqs:us-east-1:644160558196:bar_dlq",
+                "Tags": []
+            },
+            {
+                "ResourceARN": "arn:aws:sqs:us-east-1:644160558196:test",
+                "Tags": [
+                    {
+                        "Key": "Foo",
+                        "Value": "bar"
+                    }
+                ]
+            },
+            {
+                "ResourceARN": "arn:aws:sqs:us-east-1:644160558196:test-dlq",
+                "Tags": [
+                    {
+                        "Key": "Foo",
+                        "Value": "bar"
+                    }
+                ]
+            }
+        ],
+        "ResponseMetadata": {}
+    }
+}
diff --git a/tests/test_policy.py b/tests/test_policy.py
index 9270e9504..c130141cc 100644
--- a/tests/test_policy.py
+++ b/tests/test_policy.py
@@ -570,6 +570,7 @@ class PolicyMetaLint(BaseTest):
                     "instance-age",
                     "ephemeral",
                     "instance-uptime",
+                    "dead-letter",
                 ):
                     continue
                 qk = "%s.filters.%s" % (k, n)
diff --git a/tests/test_sqs.py b/tests/test_sqs.py
index 7ad5fa1c1..1a998c4a6 100644
--- a/tests/test_sqs.py
+++ b/tests/test_sqs.py
@@ -6,13 +6,14 @@ from dateutil.tz import tzutc
 from pytest_terraform import terraform
 from botocore.exceptions import ClientError
 
-
 import json
 import logging
 import pytest
 import time
 from pathlib import Path
 
+from c7n.executor import MainThreadExecutor
+from c7n.resources.sqs import SQS
 from c7n.resources.aws import shape_validate, Arn
 
 
@@ -694,3 +695,27 @@ class QueueTests(BaseTest):
         resources = p.run()
         self.assertEqual(len(resources), 1)
         self.assertIn('c7n:AccessAnalysis', resources[0])
+
+    def test_sqs_deadletter_filter(self):
+        factory = self.replay_flight_data("test_sqs_deadletter_filter")
+        self.patch(SQS, "executor_factory", MainThreadExecutor)
+        policy = self.load_policy(
+            {
+                "name": "sqs-deadletter",
+                "resource": "aws.sqs",
+                "filters": [
+                    {
+                        "type": "value",
+                        "key": "QueueArn",
+                        "value": "arn:aws:sqs:us-east-1:644160558196:bar",
+                        "op": "ne"
+                    },
+                    {
+                        "type": "dead-letter"
+                    },
+                ]
+            },
+            session_factory=factory
+        )
+        resources = policy.run()
+        self.assertEqual(len(resources), 2)