We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Ability to manage RDS DB snapshots
I'd love to be able to manage RDS DB snapshots as we do EBS snapshots:
- A new resource type (e.g. `rds-snapshot`)
- An associated `age` filter
- An associated `delete` action


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `c7n/resources/rds.py`
Content:
```
1 # Copyright 2016 Capital One Services, LLC
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 # http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 """
15 RDS Resource Manager
16 ====================
17 
18 Example Policies
19 ----------------
20 
21 Find rds instances that are publicly available
22 
23 .. code-block:: yaml
24 
25    policies:
26       - name: rds-public
27         resource: rds
28         filters:
29          - PubliclyAccessible: true
30 
31 Find rds instances that are not encrypted
32 
33 .. code-block:: yaml
34 
35    policies:
36       - name: rds-non-encrypted
37         resource: rds
38         filters:
39          - type: value
40            key: StorageEncrypted
41            value: true
42            op: ne
43 
44 
45 Todo/Notes
46 ----------
47 - Tag api for rds is highly inconsistent
48   compared to every other aws api, it
49   requires full arns. The api never exposes
50   arn. We should use a policy attribute
51   for arn, that can dereference from assume
52   role, instance profile role, iam user (GetUser),
53   or for sts assume role users we need to
54   require cli params for this resource type.
55 
56 - aurora databases also generate clusters
57   that are listed separately and return
58   different metadata using the cluster api
59 
60 
61 """
62 import logging
63 
64 from botocore.exceptions import ClientError
65 from concurrent.futures import as_completed
66 
67 from c7n.actions import ActionRegistry, BaseAction
68 from c7n.filters import FilterRegistry, Filter
69 from c7n.manager import resources
70 from c7n.query import QueryResourceManager
71 from c7n import tags
72 from c7n.utils import local_session, type_schema, get_account_id
73 
74 from skew.resources.aws import rds
75 
76 log = logging.getLogger('custodian.rds')
77 
78 filters = FilterRegistry('rds.filters')
79 actions = ActionRegistry('rds.actions')
80 
81 filters.register('tag-count', tags.TagCountFilter)
82 filters.register('marked-for-op', tags.TagActionFilter)
83 
84 
85 @resources.register('rds')
86 class RDS(QueryResourceManager):
87 
88     class resource_type(rds.DBInstance.Meta):
89         filter_name = 'DBInstanceIdentifier'
90 
91     filter_registry = filters
92     action_registry = actions
93     account_id = None
94 
95     def augment(self, resources):
96         session = local_session(self.session_factory)
97         if self.account_id is None:
98             self.account_id = get_account_id(session)
99         _rds_tags(
100             self.query.resolve(self.resource_type),
101             resources, self.session_factory, self.executor_factory,
102             self.account_id, region=self.config.region)
103 
104 
105 def _rds_tags(
106         model, dbs, session_factory, executor_factory, account_id, region):
107     """Augment rds instances with their respective tags."""
108 
109     def process_tags(db):
110         client = local_session(session_factory).client('rds')
111         arn = "arn:aws:rds:%s:%s:db:%s" % (region, account_id, db[model.id])
112         tag_list = client.list_tags_for_resource(ResourceName=arn)['TagList']
113         db['Tags'] = tag_list or []
114         return db
115 
116     # Rds maintains a low api call limit, so this can take some time :-(
117     with executor_factory(max_workers=2) as w:
118         list(w.map(process_tags, dbs))
119 
120 
121 @filters.register('default-vpc')
122 class DefaultVpc(Filter):
123     """ Matches if an rds database is in the default vpc
124     """
125 
126     schema = type_schema('default-vpc')
127 
128     vpcs = None
129     default_vpc = None
130 
131     def __call__(self, rdb):
132         vpc_id = rdb['DBSubnetGroup']['VpcId']
133         if self.vpcs is None:
134             self.vpcs = set((vpc_id,))
135             query_vpc = vpc_id
136         else:
137             query_vpc = vpc_id not in self.vpcs and vpc_id or None
138 
139         if query_vpc:
140             client = local_session(self.manager.session_factory).client('ec2')
141             self.log.debug("querying vpc %s" % vpc_id)
142             vpcs = [v['VpcId'] for v
143                     in client.describe_vpcs(VpcIds=[vpc_id])['Vpcs']
144                     if v['IsDefault']]
145             self.vpcs.add(vpc_id)
146             if not vpcs:
147                 return []
148             self.default_vpc = vpcs.pop()
149         return vpc_id == self.default_vpc and True or False
150 
151 
152 @actions.register('mark-for-op')
153 class TagDelayedAction(tags.TagDelayedAction):
154 
155     schema = type_schema(
156         'mark-for-op', rinherit=tags.TagDelayedAction.schema,
157         ops={'enum': ['delete', 'snapshot']})
158 
159     batch_size = 5
160 
161     def process(self, resources):
162         session = local_session(self.manager.session_factory)
163         return super(TagDelayedAction, self).process(resources)
164 
165     def process_resource_set(self, resources, tags):
166         client = local_session(self.manager.session_factory).client('rds')
167         for r in resources:
168             arn = "arn:aws:rds:%s:%s:db:%s" % (
169                 self.manager.config.region, self.manager.account_id,
170                 r['DBInstanceIdentifier'])
171             client.add_tags_to_resource(ResourceName=arn, Tags=tags)
172 
173 
174 @actions.register('tag')
175 class Tag(tags.Tag):
176 
177     concurrency = 2
178     batch_size = 5
179 
180     def process_resource_set(self, resources, tags):
181         client = local_session(
182             self.manager.session_factory).client('rds')
183         for r in resources:
184             arn = "arn:aws:rds:%s:%s:db:%s" % (
185                 self.manager.config.region, self.manager.account_id,
186                 r['DBInstanceIdentifier'])
187             client.add_tags_to_resource(ResourceName=arn, Tags=tags)
188 
189 
190 @actions.register('remove-tag')
191 class RemoveTag(tags.RemoveTag):
192 
193     concurrency = 2
194     batch_size = 5
195 
196     def process_resource_set(self, resources, tag_keys):
197         client = local_session(
198             self.manager.session_factory).client('rds')
199         for r in resources:
200             arn = "arn:aws:rds:%s:%s:db:%s" % (
201                 self.manager.config.region, self.manager.account_id,
202                 r['DBInstanceIdentifier'])
203             client.remove_tags_from_resource(
204                 ResourceName=arn, TagKeys=tag_keys)
205 
206 
207 @actions.register('delete')
208 class Delete(BaseAction):
209 
210     schema = {
211         'type': 'object',
212         'properties': {
213             'type': {'enum': ['delete'],
214                      'skip-snapshot': {'type': 'boolean'}}
215             }
216         }
217 
218     def process(self, resources):
219         self.skip = self.data.get('skip-snapshot', False)
220 
221         # Concurrency feels like over kill here.
222         client = local_session(self.manager.session_factory).client('rds')
223 
224         for rdb in resources:
225             params = dict(
226                 DBInstanceIdentifier=rdb['DBInstanceIdentifier'])
227             if self.skip:
228                 params['SkipFinalSnapshot'] = True
229             else:
230                 params[
231                     'FinalDBSnapshotIdentifier'] = rdb['DBInstanceIdentifier']
232             try:
233                 client.delete_db_instance(**params)
234             except ClientError as e:
235                 if e.response['Error']['Code'] == "InvalidDBInstanceState":
236                     continue
237                 raise
238 
239             self.log.info("Deleted rds: %s" % rdb['DBInstanceIdentifier'])
240 
241 
242 @actions.register('snapshot')
243 class Snapshot(BaseAction):
244 
245     schema = {'properties': {
246         'type': {
247             'enum': ['snapshot']}}}
248 
249     def process(self, resources):
250         with self.executor_factory(max_workers=3) as w:
251             futures = []
252             for resource in resources:
253                 futures.append(w.submit(
254                     self.process_rds_snapshot,
255                     resource))
256                 for f in as_completed(futures):
257                     if f.exception():
258                         self.log.error(
259                             "Exception creating rds snapshot  \n %s" % (
260                                 f.exception()))
261         return resources
262 
263     def process_rds_snapshot(self, resource):
264         c = local_session(self.manager.session_factory).client('rds')
265         c.create_db_snapshot(
266             DBSnapshotIdentifier="Backup-%s-%s" % (
267                 resource['DBInstanceIdentifier'],
268                 resource['Engine']),
269             DBInstanceIdentifier=resource['DBInstanceIdentifier'])
270 
271 
272 @actions.register('retention')
273 class RetentionWindow(BaseAction):
274 
275     date_attribute = "BackupRetentionPeriod"
276     schema = type_schema('retention', days={'type': 'number'})
277 
278     def process(self, resources):
279         with self.executor_factory(max_workers=3) as w:
280             futures = []
281             for resource in resources:
282                 futures.append(w.submit(
283                     self.process_snapshot_retention,
284                     resource))
285                 for f in as_completed(futures):
286                     if f.exception():
287                         self.log.error(
288                             "Exception setting rds retention  \n %s" % (
289                                 f.exception()))
290 
291     def process_snapshot_retention(self, resource):
292         v = int(resource.get('BackupRetentionPeriod', 0))
293         if v == 0 or v < self.data['days']:
294             self.set_retention_window(resource)
295             return resource
296 
297     def set_retention_window(self, resource):
298         c = local_session(self.manager.session_factory).client('rds')
299         c.modify_db_instance(
300             DBInstanceIdentifier=resource['DBInstanceIdentifier'],
301             BackupRetentionPeriod=self.data['days'])
302
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/c7n/resources/rds.py b/c7n/resources/rds.py
--- a/c7n/resources/rds.py
+++ b/c7n/resources/rds.py
@@ -65,11 +65,11 @@
 from concurrent.futures import as_completed
 
 from c7n.actions import ActionRegistry, BaseAction
-from c7n.filters import FilterRegistry, Filter
+from c7n.filters import FilterRegistry, Filter, AgeFilter
 from c7n.manager import resources
 from c7n.query import QueryResourceManager
 from c7n import tags
-from c7n.utils import local_session, type_schema, get_account_id
+from c7n.utils import local_session, type_schema, get_account_id, chunks
 
 from skew.resources.aws import rds
 
@@ -299,3 +299,56 @@
         c.modify_db_instance(
             DBInstanceIdentifier=resource['DBInstanceIdentifier'],
             BackupRetentionPeriod=self.data['days'])
+
+
+@resources.register('rds-snapshot')
+class RDSSnapshot(QueryResourceManager):
+
+    class Meta(object):
+
+        service = 'rds'
+        type = 'rds-snapshot'
+        enum_spec = ('describe_db_snapshots', 'DBSnapshots', None)
+        name = id = 'DBSnapshotIdentifier'
+        filter_name = None
+        filter_type = None
+        dimension = None
+        date = 'SnapshotCreateTime'
+
+    resource_type = Meta
+
+    filter_registry = FilterRegistry('rds-snapshot.filters')
+    action_registry = ActionRegistry('rds-snapshot.actions')
+
+
+@RDSSnapshot.filter_registry.register('age')
+class RDSSnapshotAge(AgeFilter):
+
+    schema = type_schema('age', days={'type': 'number'})
+    date_attribute = 'SnapshotCreateTime'
+
+@RDSSnapshot.action_registry.register('delete')
+class RDSSnapshotDelete(BaseAction):
+
+    def process(self, snapshots):
+        log.info("Deleting %d rds snapshots", len(snapshots))
+        with self.executor_factory(max_workers=3) as w:
+            futures = []
+            for snapshot_set in chunks(reversed(snapshots), size=50):
+                futures.append(
+                    w.submit(self.process_snapshot_set, snapshot_set))
+                for f in as_completed(futures):
+                    if f.exception():
+                        self.log.error(
+                            "Exception deleting snapshot set \n %s" % (
+                                f.exception()))
+        return snapshots
+
+    def process_snapshot_set(self, snapshots_set):
+        c = local_session(self.manager.session_factory).client('rds')
+        for s in snapshots_set:
+            try:
+                c.delete_db_snapshot(
+                    DBSnapshotIdentifier=s['DBSnapshotIdentifier'])
+            except ClientError as e:
+                raise

{"golden_diff": "diff --git a/c7n/resources/rds.py b/c7n/resources/rds.py\n--- a/c7n/resources/rds.py\n+++ b/c7n/resources/rds.py\n@@ -65,11 +65,11 @@\n from concurrent.futures import as_completed\n \n from c7n.actions import ActionRegistry, BaseAction\n-from c7n.filters import FilterRegistry, Filter\n+from c7n.filters import FilterRegistry, Filter, AgeFilter\n from c7n.manager import resources\n from c7n.query import QueryResourceManager\n from c7n import tags\n-from c7n.utils import local_session, type_schema, get_account_id\n+from c7n.utils import local_session, type_schema, get_account_id, chunks\n \n from skew.resources.aws import rds\n \n@@ -299,3 +299,56 @@\n         c.modify_db_instance(\n             DBInstanceIdentifier=resource['DBInstanceIdentifier'],\n             BackupRetentionPeriod=self.data['days'])\n+\n+\n+@resources.register('rds-snapshot')\n+class RDSSnapshot(QueryResourceManager):\n+\n+    class Meta(object):\n+\n+        service = 'rds'\n+        type = 'rds-snapshot'\n+        enum_spec = ('describe_db_snapshots', 'DBSnapshots', None)\n+        name = id = 'DBSnapshotIdentifier'\n+        filter_name = None\n+        filter_type = None\n+        dimension = None\n+        date = 'SnapshotCreateTime'\n+\n+    resource_type = Meta\n+\n+    filter_registry = FilterRegistry('rds-snapshot.filters')\n+    action_registry = ActionRegistry('rds-snapshot.actions')\n+\n+\n+@RDSSnapshot.filter_registry.register('age')\n+class RDSSnapshotAge(AgeFilter):\n+\n+    schema = type_schema('age', days={'type': 'number'})\n+    date_attribute = 'SnapshotCreateTime'\n+\n+@RDSSnapshot.action_registry.register('delete')\n+class RDSSnapshotDelete(BaseAction):\n+\n+    def process(self, snapshots):\n+        log.info(\"Deleting %d rds snapshots\", len(snapshots))\n+        with self.executor_factory(max_workers=3) as w:\n+            futures = []\n+            for snapshot_set in chunks(reversed(snapshots), size=50):\n+                futures.append(\n+                    w.submit(self.process_snapshot_set, snapshot_set))\n+                for f in as_completed(futures):\n+                    if f.exception():\n+                        self.log.error(\n+                            \"Exception deleting snapshot set \\n %s\" % (\n+                                f.exception()))\n+        return snapshots\n+\n+    def process_snapshot_set(self, snapshots_set):\n+        c = local_session(self.manager.session_factory).client('rds')\n+        for s in snapshots_set:\n+            try:\n+                c.delete_db_snapshot(\n+                    DBSnapshotIdentifier=s['DBSnapshotIdentifier'])\n+            except ClientError as e:\n+                raise\n", "issue": "Ability to manage RDS DB snapshots\nI'd love to be able to manage RDS DB snapshots as we do EBS snapshots:\n- A new resource type (e.g. `rds-snapshot`)\n- An associated `age` filter\n- An associated `delete` action\n\n", "before_files": [{"content": "# Copyright 2016 Capital One Services, LLC\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"\nRDS Resource Manager\n====================\n\nExample Policies\n----------------\n\nFind rds instances that are publicly available\n\n.. code-block:: yaml\n\n   policies:\n      - name: rds-public\n        resource: rds\n        filters:\n         - PubliclyAccessible: true\n\nFind rds instances that are not encrypted\n\n.. code-block:: yaml\n\n   policies:\n      - name: rds-non-encrypted\n        resource: rds\n        filters:\n         - type: value\n           key: StorageEncrypted\n           value: true\n           op: ne\n\n\nTodo/Notes\n----------\n- Tag api for rds is highly inconsistent\n  compared to every other aws api, it\n  requires full arns. The api never exposes\n  arn. We should use a policy attribute\n  for arn, that can dereference from assume\n  role, instance profile role, iam user (GetUser),\n  or for sts assume role users we need to\n  require cli params for this resource type.\n\n- aurora databases also generate clusters\n  that are listed separately and return\n  different metadata using the cluster api\n\n\n\"\"\"\nimport logging\n\nfrom botocore.exceptions import ClientError\nfrom concurrent.futures import as_completed\n\nfrom c7n.actions import ActionRegistry, BaseAction\nfrom c7n.filters import FilterRegistry, Filter\nfrom c7n.manager import resources\nfrom c7n.query import QueryResourceManager\nfrom c7n import tags\nfrom c7n.utils import local_session, type_schema, get_account_id\n\nfrom skew.resources.aws import rds\n\nlog = logging.getLogger('custodian.rds')\n\nfilters = FilterRegistry('rds.filters')\nactions = ActionRegistry('rds.actions')\n\nfilters.register('tag-count', tags.TagCountFilter)\nfilters.register('marked-for-op', tags.TagActionFilter)\n\n\n@resources.register('rds')\nclass RDS(QueryResourceManager):\n\n    class resource_type(rds.DBInstance.Meta):\n        filter_name = 'DBInstanceIdentifier'\n\n    filter_registry = filters\n    action_registry = actions\n    account_id = None\n\n    def augment(self, resources):\n        session = local_session(self.session_factory)\n        if self.account_id is None:\n            self.account_id = get_account_id(session)\n        _rds_tags(\n            self.query.resolve(self.resource_type),\n            resources, self.session_factory, self.executor_factory,\n            self.account_id, region=self.config.region)\n\n\ndef _rds_tags(\n        model, dbs, session_factory, executor_factory, account_id, region):\n    \"\"\"Augment rds instances with their respective tags.\"\"\"\n\n    def process_tags(db):\n        client = local_session(session_factory).client('rds')\n        arn = \"arn:aws:rds:%s:%s:db:%s\" % (region, account_id, db[model.id])\n        tag_list = client.list_tags_for_resource(ResourceName=arn)['TagList']\n        db['Tags'] = tag_list or []\n        return db\n\n    # Rds maintains a low api call limit, so this can take some time :-(\n    with executor_factory(max_workers=2) as w:\n        list(w.map(process_tags, dbs))\n\n\n@filters.register('default-vpc')\nclass DefaultVpc(Filter):\n    \"\"\" Matches if an rds database is in the default vpc\n    \"\"\"\n\n    schema = type_schema('default-vpc')\n\n    vpcs = None\n    default_vpc = None\n\n    def __call__(self, rdb):\n        vpc_id = rdb['DBSubnetGroup']['VpcId']\n        if self.vpcs is None:\n            self.vpcs = set((vpc_id,))\n            query_vpc = vpc_id\n        else:\n            query_vpc = vpc_id not in self.vpcs and vpc_id or None\n\n        if query_vpc:\n            client = local_session(self.manager.session_factory).client('ec2')\n            self.log.debug(\"querying vpc %s\" % vpc_id)\n            vpcs = [v['VpcId'] for v\n                    in client.describe_vpcs(VpcIds=[vpc_id])['Vpcs']\n                    if v['IsDefault']]\n            self.vpcs.add(vpc_id)\n            if not vpcs:\n                return []\n            self.default_vpc = vpcs.pop()\n        return vpc_id == self.default_vpc and True or False\n\n\n@actions.register('mark-for-op')\nclass TagDelayedAction(tags.TagDelayedAction):\n\n    schema = type_schema(\n        'mark-for-op', rinherit=tags.TagDelayedAction.schema,\n        ops={'enum': ['delete', 'snapshot']})\n\n    batch_size = 5\n\n    def process(self, resources):\n        session = local_session(self.manager.session_factory)\n        return super(TagDelayedAction, self).process(resources)\n\n    def process_resource_set(self, resources, tags):\n        client = local_session(self.manager.session_factory).client('rds')\n        for r in resources:\n            arn = \"arn:aws:rds:%s:%s:db:%s\" % (\n                self.manager.config.region, self.manager.account_id,\n                r['DBInstanceIdentifier'])\n            client.add_tags_to_resource(ResourceName=arn, Tags=tags)\n\n\n@actions.register('tag')\nclass Tag(tags.Tag):\n\n    concurrency = 2\n    batch_size = 5\n\n    def process_resource_set(self, resources, tags):\n        client = local_session(\n            self.manager.session_factory).client('rds')\n        for r in resources:\n            arn = \"arn:aws:rds:%s:%s:db:%s\" % (\n                self.manager.config.region, self.manager.account_id,\n                r['DBInstanceIdentifier'])\n            client.add_tags_to_resource(ResourceName=arn, Tags=tags)\n\n\n@actions.register('remove-tag')\nclass RemoveTag(tags.RemoveTag):\n\n    concurrency = 2\n    batch_size = 5\n\n    def process_resource_set(self, resources, tag_keys):\n        client = local_session(\n            self.manager.session_factory).client('rds')\n        for r in resources:\n            arn = \"arn:aws:rds:%s:%s:db:%s\" % (\n                self.manager.config.region, self.manager.account_id,\n                r['DBInstanceIdentifier'])\n            client.remove_tags_from_resource(\n                ResourceName=arn, TagKeys=tag_keys)\n\n\n@actions.register('delete')\nclass Delete(BaseAction):\n\n    schema = {\n        'type': 'object',\n        'properties': {\n            'type': {'enum': ['delete'],\n                     'skip-snapshot': {'type': 'boolean'}}\n            }\n        }\n\n    def process(self, resources):\n        self.skip = self.data.get('skip-snapshot', False)\n\n        # Concurrency feels like over kill here.\n        client = local_session(self.manager.session_factory).client('rds')\n\n        for rdb in resources:\n            params = dict(\n                DBInstanceIdentifier=rdb['DBInstanceIdentifier'])\n            if self.skip:\n                params['SkipFinalSnapshot'] = True\n            else:\n                params[\n                    'FinalDBSnapshotIdentifier'] = rdb['DBInstanceIdentifier']\n            try:\n                client.delete_db_instance(**params)\n            except ClientError as e:\n                if e.response['Error']['Code'] == \"InvalidDBInstanceState\":\n                    continue\n                raise\n\n            self.log.info(\"Deleted rds: %s\" % rdb['DBInstanceIdentifier'])\n\n\n@actions.register('snapshot')\nclass Snapshot(BaseAction):\n\n    schema = {'properties': {\n        'type': {\n            'enum': ['snapshot']}}}\n\n    def process(self, resources):\n        with self.executor_factory(max_workers=3) as w:\n            futures = []\n            for resource in resources:\n                futures.append(w.submit(\n                    self.process_rds_snapshot,\n                    resource))\n                for f in as_completed(futures):\n                    if f.exception():\n                        self.log.error(\n                            \"Exception creating rds snapshot  \\n %s\" % (\n                                f.exception()))\n        return resources\n\n    def process_rds_snapshot(self, resource):\n        c = local_session(self.manager.session_factory).client('rds')\n        c.create_db_snapshot(\n            DBSnapshotIdentifier=\"Backup-%s-%s\" % (\n                resource['DBInstanceIdentifier'],\n                resource['Engine']),\n            DBInstanceIdentifier=resource['DBInstanceIdentifier'])\n\n\n@actions.register('retention')\nclass RetentionWindow(BaseAction):\n\n    date_attribute = \"BackupRetentionPeriod\"\n    schema = type_schema('retention', days={'type': 'number'})\n\n    def process(self, resources):\n        with self.executor_factory(max_workers=3) as w:\n            futures = []\n            for resource in resources:\n                futures.append(w.submit(\n                    self.process_snapshot_retention,\n                    resource))\n                for f in as_completed(futures):\n                    if f.exception():\n                        self.log.error(\n                            \"Exception setting rds retention  \\n %s\" % (\n                                f.exception()))\n\n    def process_snapshot_retention(self, resource):\n        v = int(resource.get('BackupRetentionPeriod', 0))\n        if v == 0 or v < self.data['days']:\n            self.set_retention_window(resource)\n            return resource\n\n    def set_retention_window(self, resource):\n        c = local_session(self.manager.session_factory).client('rds')\n        c.modify_db_instance(\n            DBInstanceIdentifier=resource['DBInstanceIdentifier'],\n            BackupRetentionPeriod=self.data['days'])\n", "path": "c7n/resources/rds.py"}], "after_files": [{"content": "# Copyright 2016 Capital One Services, LLC\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"\nRDS Resource Manager\n====================\n\nExample Policies\n----------------\n\nFind rds instances that are publicly available\n\n.. code-block:: yaml\n\n   policies:\n      - name: rds-public\n        resource: rds\n        filters:\n         - PubliclyAccessible: true\n\nFind rds instances that are not encrypted\n\n.. code-block:: yaml\n\n   policies:\n      - name: rds-non-encrypted\n        resource: rds\n        filters:\n         - type: value\n           key: StorageEncrypted\n           value: true\n           op: ne\n\n\nTodo/Notes\n----------\n- Tag api for rds is highly inconsistent\n  compared to every other aws api, it\n  requires full arns. The api never exposes\n  arn. We should use a policy attribute\n  for arn, that can dereference from assume\n  role, instance profile role, iam user (GetUser),\n  or for sts assume role users we need to\n  require cli params for this resource type.\n\n- aurora databases also generate clusters\n  that are listed separately and return\n  different metadata using the cluster api\n\n\n\"\"\"\nimport logging\n\nfrom botocore.exceptions import ClientError\nfrom concurrent.futures import as_completed\n\nfrom c7n.actions import ActionRegistry, BaseAction\nfrom c7n.filters import FilterRegistry, Filter, AgeFilter\nfrom c7n.manager import resources\nfrom c7n.query import QueryResourceManager\nfrom c7n import tags\nfrom c7n.utils import local_session, type_schema, get_account_id, chunks\n\nfrom skew.resources.aws import rds\n\nlog = logging.getLogger('custodian.rds')\n\nfilters = FilterRegistry('rds.filters')\nactions = ActionRegistry('rds.actions')\n\nfilters.register('tag-count', tags.TagCountFilter)\nfilters.register('marked-for-op', tags.TagActionFilter)\n\n\n@resources.register('rds')\nclass RDS(QueryResourceManager):\n\n    class resource_type(rds.DBInstance.Meta):\n        filter_name = 'DBInstanceIdentifier'\n\n    filter_registry = filters\n    action_registry = actions\n    account_id = None\n\n    def augment(self, resources):\n        session = local_session(self.session_factory)\n        if self.account_id is None:\n            self.account_id = get_account_id(session)\n        _rds_tags(\n            self.query.resolve(self.resource_type),\n            resources, self.session_factory, self.executor_factory,\n            self.account_id, region=self.config.region)\n\n\ndef _rds_tags(\n        model, dbs, session_factory, executor_factory, account_id, region):\n    \"\"\"Augment rds instances with their respective tags.\"\"\"\n\n    def process_tags(db):\n        client = local_session(session_factory).client('rds')\n        arn = \"arn:aws:rds:%s:%s:db:%s\" % (region, account_id, db[model.id])\n        tag_list = client.list_tags_for_resource(ResourceName=arn)['TagList']\n        db['Tags'] = tag_list or []\n        return db\n\n    # Rds maintains a low api call limit, so this can take some time :-(\n    with executor_factory(max_workers=2) as w:\n        list(w.map(process_tags, dbs))\n\n\n@filters.register('default-vpc')\nclass DefaultVpc(Filter):\n    \"\"\" Matches if an rds database is in the default vpc\n    \"\"\"\n\n    schema = type_schema('default-vpc')\n\n    vpcs = None\n    default_vpc = None\n\n    def __call__(self, rdb):\n        vpc_id = rdb['DBSubnetGroup']['VpcId']\n        if self.vpcs is None:\n            self.vpcs = set((vpc_id,))\n            query_vpc = vpc_id\n        else:\n            query_vpc = vpc_id not in self.vpcs and vpc_id or None\n\n        if query_vpc:\n            client = local_session(self.manager.session_factory).client('ec2')\n            self.log.debug(\"querying vpc %s\" % vpc_id)\n            vpcs = [v['VpcId'] for v\n                    in client.describe_vpcs(VpcIds=[vpc_id])['Vpcs']\n                    if v['IsDefault']]\n            self.vpcs.add(vpc_id)\n            if not vpcs:\n                return []\n            self.default_vpc = vpcs.pop()\n        return vpc_id == self.default_vpc and True or False\n\n\n@actions.register('mark-for-op')\nclass TagDelayedAction(tags.TagDelayedAction):\n\n    schema = type_schema(\n        'mark-for-op', rinherit=tags.TagDelayedAction.schema,\n        ops={'enum': ['delete', 'snapshot']})\n\n    batch_size = 5\n\n    def process(self, resources):\n        session = local_session(self.manager.session_factory)\n        return super(TagDelayedAction, self).process(resources)\n\n    def process_resource_set(self, resources, tags):\n        client = local_session(self.manager.session_factory).client('rds')\n        for r in resources:\n            arn = \"arn:aws:rds:%s:%s:db:%s\" % (\n                self.manager.config.region, self.manager.account_id,\n                r['DBInstanceIdentifier'])\n            client.add_tags_to_resource(ResourceName=arn, Tags=tags)\n\n\n@actions.register('tag')\nclass Tag(tags.Tag):\n\n    concurrency = 2\n    batch_size = 5\n\n    def process_resource_set(self, resources, tags):\n        client = local_session(\n            self.manager.session_factory).client('rds')\n        for r in resources:\n            arn = \"arn:aws:rds:%s:%s:db:%s\" % (\n                self.manager.config.region, self.manager.account_id,\n                r['DBInstanceIdentifier'])\n            client.add_tags_to_resource(ResourceName=arn, Tags=tags)\n\n\n@actions.register('remove-tag')\nclass RemoveTag(tags.RemoveTag):\n\n    concurrency = 2\n    batch_size = 5\n\n    def process_resource_set(self, resources, tag_keys):\n        client = local_session(\n            self.manager.session_factory).client('rds')\n        for r in resources:\n            arn = \"arn:aws:rds:%s:%s:db:%s\" % (\n                self.manager.config.region, self.manager.account_id,\n                r['DBInstanceIdentifier'])\n            client.remove_tags_from_resource(\n                ResourceName=arn, TagKeys=tag_keys)\n\n\n@actions.register('delete')\nclass Delete(BaseAction):\n\n    schema = {\n        'type': 'object',\n        'properties': {\n            'type': {'enum': ['delete'],\n                     'skip-snapshot': {'type': 'boolean'}}\n            }\n        }\n\n    def process(self, resources):\n        self.skip = self.data.get('skip-snapshot', False)\n\n        # Concurrency feels like over kill here.\n        client = local_session(self.manager.session_factory).client('rds')\n\n        for rdb in resources:\n            params = dict(\n                DBInstanceIdentifier=rdb['DBInstanceIdentifier'])\n            if self.skip:\n                params['SkipFinalSnapshot'] = True\n            else:\n                params[\n                    'FinalDBSnapshotIdentifier'] = rdb['DBInstanceIdentifier']\n            try:\n                client.delete_db_instance(**params)\n            except ClientError as e:\n                if e.response['Error']['Code'] == \"InvalidDBInstanceState\":\n                    continue\n                raise\n\n            self.log.info(\"Deleted rds: %s\" % rdb['DBInstanceIdentifier'])\n\n\n@actions.register('snapshot')\nclass Snapshot(BaseAction):\n\n    schema = {'properties': {\n        'type': {\n            'enum': ['snapshot']}}}\n\n    def process(self, resources):\n        with self.executor_factory(max_workers=3) as w:\n            futures = []\n            for resource in resources:\n                futures.append(w.submit(\n                    self.process_rds_snapshot,\n                    resource))\n                for f in as_completed(futures):\n                    if f.exception():\n                        self.log.error(\n                            \"Exception creating rds snapshot  \\n %s\" % (\n                                f.exception()))\n        return resources\n\n    def process_rds_snapshot(self, resource):\n        c = local_session(self.manager.session_factory).client('rds')\n        c.create_db_snapshot(\n            DBSnapshotIdentifier=\"Backup-%s-%s\" % (\n                resource['DBInstanceIdentifier'],\n                resource['Engine']),\n            DBInstanceIdentifier=resource['DBInstanceIdentifier'])\n\n\n@actions.register('retention')\nclass RetentionWindow(BaseAction):\n\n    date_attribute = \"BackupRetentionPeriod\"\n    schema = type_schema('retention', days={'type': 'number'})\n\n    def process(self, resources):\n        with self.executor_factory(max_workers=3) as w:\n            futures = []\n            for resource in resources:\n                futures.append(w.submit(\n                    self.process_snapshot_retention,\n                    resource))\n                for f in as_completed(futures):\n                    if f.exception():\n                        self.log.error(\n                            \"Exception setting rds retention  \\n %s\" % (\n                                f.exception()))\n\n    def process_snapshot_retention(self, resource):\n        v = int(resource.get('BackupRetentionPeriod', 0))\n        if v == 0 or v < self.data['days']:\n            self.set_retention_window(resource)\n            return resource\n\n    def set_retention_window(self, resource):\n        c = local_session(self.manager.session_factory).client('rds')\n        c.modify_db_instance(\n            DBInstanceIdentifier=resource['DBInstanceIdentifier'],\n            BackupRetentionPeriod=self.data['days'])\n\n\n@resources.register('rds-snapshot')\nclass RDSSnapshot(QueryResourceManager):\n\n    class Meta(object):\n\n        service = 'rds'\n        type = 'rds-snapshot'\n        enum_spec = ('describe_db_snapshots', 'DBSnapshots', None)\n        name = id = 'DBSnapshotIdentifier'\n        filter_name = None\n        filter_type = None\n        dimension = None\n        date = 'SnapshotCreateTime'\n\n    resource_type = Meta\n\n    filter_registry = FilterRegistry('rds-snapshot.filters')\n    action_registry = ActionRegistry('rds-snapshot.actions')\n\n\n@RDSSnapshot.filter_registry.register('age')\nclass RDSSnapshotAge(AgeFilter):\n\n    schema = type_schema('age', days={'type': 'number'})\n    date_attribute = 'SnapshotCreateTime'\n\n@RDSSnapshot.action_registry.register('delete')\nclass RDSSnapshotDelete(BaseAction):\n\n    def process(self, snapshots):\n        log.info(\"Deleting %d rds snapshots\", len(snapshots))\n        with self.executor_factory(max_workers=3) as w:\n            futures = []\n            for snapshot_set in chunks(reversed(snapshots), size=50):\n                futures.append(\n                    w.submit(self.process_snapshot_set, snapshot_set))\n                for f in as_completed(futures):\n                    if f.exception():\n                        self.log.error(\n                            \"Exception deleting snapshot set \\n %s\" % (\n                                f.exception()))\n        return snapshots\n\n    def process_snapshot_set(self, snapshots_set):\n        c = local_session(self.manager.session_factory).client('rds')\n        for s in snapshots_set:\n            try:\n                c.delete_db_snapshot(\n                    DBSnapshotIdentifier=s['DBSnapshotIdentifier'])\n            except ClientError as e:\n                raise\n", "path": "c7n/resources/rds.py"}]}