We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Marking emr cluster for termination throws exception
When marking EMR cluster for termination throws exception in version c7n-0.8.31.2. I see the tag created in cluster
`````
policies:
- name: emr-mark-clusters-for-termination
  resource: emr
  filters:
    - type: value
      key: "Id"
      op: in
      value:
        - 'abcdefghij'
  actions:
    - type: mark-for-op
      tag: 'custodian-emr-terminate'
      op: terminate
      days: 4
`````

this policy throws exception 

2018-09-27 19:20:30,262: custodian.actions:INFO Tagging 1 resources for terminate on 2018/10/01
2018-09-27 19:20:31,720: custodian.actions:ERROR Exception with tags: [{u'Value': u'Resource does not meet policy: terminate@2018/10/01', u'Key': 'custodian-emr-terminate'}] on resources: abcdefghij
 'dict' object is not callable
`
Though the EMR is marked with tag ''custodian-emr-terminate', filtering on type: marked-for-op, returns 0 resources.

--- 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/emr.py`
Content:
```
1 # Copyright 2016-2017 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 from __future__ import absolute_import, division, print_function, unicode_literals
15 
16 import logging
17 import time
18 
19 import six
20 
21 from c7n.actions import ActionRegistry, BaseAction
22 from c7n.exceptions import PolicyValidationError
23 from c7n.filters import FilterRegistry, MetricsFilter
24 from c7n.manager import resources
25 from c7n.query import QueryResourceManager
26 from c7n.utils import (
27     local_session, type_schema, get_retry)
28 from c7n.tags import (
29     TagDelayedAction, RemoveTag, TagActionFilter, Tag)
30 
31 filters = FilterRegistry('emr.filters')
32 actions = ActionRegistry('emr.actions')
33 log = logging.getLogger('custodian.emr')
34 
35 filters.register('marked-for-op', TagActionFilter)
36 
37 
38 @resources.register('emr')
39 class EMRCluster(QueryResourceManager):
40     """Resource manager for Elastic MapReduce clusters
41     """
42 
43     class resource_type(object):
44         service = 'emr'
45         type = 'emr'
46         cluster_states = ['WAITING', 'BOOTSTRAPPING', 'RUNNING', 'STARTING']
47         enum_spec = ('list_clusters', 'Clusters', {'ClusterStates': cluster_states})
48         name = 'Name'
49         id = 'Id'
50         date = "Status.Timeline.CreationDateTime"
51         filter_name = None
52         dimension = None
53 
54     action_registry = actions
55     filter_registry = filters
56     retry = staticmethod(get_retry(('ThrottlingException',)))
57 
58     def __init__(self, ctx, data):
59         super(EMRCluster, self).__init__(ctx, data)
60         self.queries = QueryFilter.parse(
61             self.data.get('query', [
62                 {'ClusterStates': [
63                     'running', 'bootstrapping', 'waiting']}]))
64 
65     @classmethod
66     def get_permissions(cls):
67         return ("elasticmapreduce:ListClusters",
68                 "elasticmapreduce:DescribeCluster")
69 
70     def get_resources(self, ids):
71         # no filtering by id set supported at the api
72         client = local_session(self.session_factory).client('emr')
73         results = []
74         for jid in ids:
75             results.append(
76                 client.describe_cluster(ClusterId=jid)['Cluster'])
77         return results
78 
79     def resources(self, query=None):
80         q = self.consolidate_query_filter()
81         if q is not None:
82             query = query or {}
83             for i in range(0, len(q)):
84                 query[q[i]['Name']] = q[i]['Values']
85         return super(EMRCluster, self).resources(query=query)
86 
87     def consolidate_query_filter(self):
88         result = []
89         names = set()
90         # allow same name to be specified multiple times and append the queries
91         # under the same name
92         for q in self.queries:
93             query_filter = q.query()
94             if query_filter['Name'] in names:
95                 for filt in result:
96                     if query_filter['Name'] == filt['Name']:
97                         filt['Values'].extend(query_filter['Values'])
98             else:
99                 names.add(query_filter['Name'])
100                 result.append(query_filter)
101         if 'ClusterStates' not in names:
102             # include default query
103             result.append(
104                 {
105                     'Name': 'ClusterStates',
106                     'Values': ['WAITING', 'RUNNING', 'BOOTSTRAPPING'],
107                 }
108             )
109         return result
110 
111     def augment(self, resources):
112         client = local_session(
113             self.get_resource_manager('emr').session_factory).client('emr')
114         result = []
115         # remap for cwmetrics
116         for r in resources:
117             cluster = self.retry(
118                 client.describe_cluster, ClusterId=r['Id'])['Cluster']
119             result.append(cluster)
120         return result
121 
122 
123 @EMRCluster.filter_registry.register('metrics')
124 class EMRMetrics(MetricsFilter):
125 
126     def get_dimensions(self, resource):
127         # Job flow id is legacy name for cluster id
128         return [{'Name': 'JobFlowId', 'Value': resource['Id']}]
129 
130 
131 @actions.register('mark-for-op')
132 class TagDelayedAction(TagDelayedAction):
133     """Action to specify an action to occur at a later date
134 
135     :example:
136 
137     .. code-block:: yaml
138 
139             policies:
140               - name: emr-mark-for-op
141                 resource: emr
142                 filters:
143                   - "tag:Name": absent
144                 actions:
145                   - type: mark-for-op
146                     tag: custodian_cleanup
147                     op: terminate
148                     days: 4
149                     msg: "Cluster does not have required tags"
150     """
151 
152     permission = ('elasticmapreduce:AddTags',)
153     batch_size = 1
154     retry = staticmethod(get_retry(('ThrottlingException',)))
155 
156     def process_resource_set(self, resources, tags):
157         client = local_session(
158             self.manager.session_factory).client('emr')
159         for r in resources:
160             self.retry(client.add_tags(ResourceId=r['Id'], Tags=tags))
161 
162 
163 @actions.register('tag')
164 class TagTable(Tag):
165     """Action to create tag(s) on a resource
166 
167     :example:
168 
169     .. code-block:: yaml
170 
171             policies:
172               - name: emr-tag-table
173                 resource: emr
174                 filters:
175                   - "tag:target-tag": absent
176                 actions:
177                   - type: tag
178                     key: target-tag
179                     value: target-tag-value
180     """
181 
182     permissions = ('elasticmapreduce:AddTags',)
183     batch_size = 1
184     retry = staticmethod(get_retry(('ThrottlingException',)))
185 
186     def process_resource_set(self, resources, tags):
187         client = local_session(self.manager.session_factory).client('emr')
188         for r in resources:
189             self.retry(client.add_tags(ResourceId=r['Id'], Tags=tags))
190 
191 
192 @actions.register('remove-tag')
193 class UntagTable(RemoveTag):
194     """Action to remove tag(s) on a resource
195 
196     :example:
197 
198     .. code-block:: yaml
199 
200             policies:
201               - name: emr-remove-tag
202                 resource: emr
203                 filters:
204                   - "tag:target-tag": present
205                 actions:
206                   - type: remove-tag
207                     tags: ["target-tag"]
208     """
209 
210     concurrency = 2
211     batch_size = 5
212     permissions = ('elasticmapreduce:RemoveTags',)
213 
214     def process_resource_set(self, resources, tag_keys):
215         client = local_session(
216             self.manager.session_factory).client('emr')
217         for r in resources:
218             client.remove_tags(
219                 ResourceId=r['Id'], TagKeys=tag_keys)
220 
221 
222 @actions.register('terminate')
223 class Terminate(BaseAction):
224     """Action to terminate EMR cluster(s)
225 
226     It is recommended to apply a filter to the terminate action to avoid
227     termination of all EMR clusters
228 
229     :example:
230 
231     .. code-block:: yaml
232 
233             policies:
234               - name: emr-terminate
235                 resource: emr
236                 query:
237                   - ClusterStates: [STARTING, BOOTSTRAPPING, RUNNING, WAITING]
238                 actions:
239                   - terminate
240     """
241 
242     schema = type_schema('terminate', force={'type': 'boolean'})
243     permissions = ("elasticmapreduce:TerminateJobFlows",)
244     delay = 5
245 
246     def process(self, emrs):
247         client = local_session(self.manager.session_factory).client('emr')
248         cluster_ids = [emr['Id'] for emr in emrs]
249         if self.data.get('force'):
250             client.set_termination_protection(
251                 JobFlowIds=cluster_ids, TerminationProtected=False)
252             time.sleep(self.delay)
253         client.terminate_job_flows(JobFlowIds=cluster_ids)
254         self.log.info("Deleted emrs: %s", cluster_ids)
255         return emrs
256 
257 
258 # Valid EMR Query Filters
259 EMR_VALID_FILTERS = set(('CreatedAfter', 'CreatedBefore', 'ClusterStates'))
260 
261 
262 class QueryFilter(object):
263 
264     @classmethod
265     def parse(cls, data):
266         results = []
267         for d in data:
268             if not isinstance(d, dict):
269                 raise PolicyValidationError(
270                     "EMR Query Filter Invalid structure %s" % d)
271             results.append(cls(d).validate())
272         return results
273 
274     def __init__(self, data):
275         self.data = data
276         self.key = None
277         self.value = None
278 
279     def validate(self):
280         if not len(list(self.data.keys())) == 1:
281             raise PolicyValidationError(
282                 "EMR Query Filter Invalid %s" % self.data)
283         self.key = list(self.data.keys())[0]
284         self.value = list(self.data.values())[0]
285 
286         if self.key not in EMR_VALID_FILTERS and not self.key.startswith(
287                 'tag:'):
288             raise PolicyValidationError(
289                 "EMR Query Filter invalid filter name %s" % (self.data))
290 
291         if self.value is None:
292             raise PolicyValidationError(
293                 "EMR Query Filters must have a value, use tag-key"
294                 " w/ tag name as value for tag present checks"
295                 " %s" % self.data)
296         return self
297 
298     def query(self):
299         value = self.value
300         if isinstance(self.value, six.string_types):
301             value = [self.value]
302 
303         return {'Name': self.key, 'Values': value}
304
```
--- 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/emr.py b/c7n/resources/emr.py
--- a/c7n/resources/emr.py
+++ b/c7n/resources/emr.py
@@ -157,7 +157,7 @@
         client = local_session(
             self.manager.session_factory).client('emr')
         for r in resources:
-            self.retry(client.add_tags(ResourceId=r['Id'], Tags=tags))
+            self.retry(client.add_tags, ResourceId=r['Id'], Tags=tags)
 
 
 @actions.register('tag')
@@ -186,7 +186,7 @@
     def process_resource_set(self, resources, tags):
         client = local_session(self.manager.session_factory).client('emr')
         for r in resources:
-            self.retry(client.add_tags(ResourceId=r['Id'], Tags=tags))
+            self.retry(client.add_tags, ResourceId=r['Id'], Tags=tags)
 
 
 @actions.register('remove-tag')

{"golden_diff": "diff --git a/c7n/resources/emr.py b/c7n/resources/emr.py\n--- a/c7n/resources/emr.py\n+++ b/c7n/resources/emr.py\n@@ -157,7 +157,7 @@\n         client = local_session(\n             self.manager.session_factory).client('emr')\n         for r in resources:\n-            self.retry(client.add_tags(ResourceId=r['Id'], Tags=tags))\n+            self.retry(client.add_tags, ResourceId=r['Id'], Tags=tags)\n \n \n @actions.register('tag')\n@@ -186,7 +186,7 @@\n     def process_resource_set(self, resources, tags):\n         client = local_session(self.manager.session_factory).client('emr')\n         for r in resources:\n-            self.retry(client.add_tags(ResourceId=r['Id'], Tags=tags))\n+            self.retry(client.add_tags, ResourceId=r['Id'], Tags=tags)\n \n \n @actions.register('remove-tag')\n", "issue": "Marking emr cluster for termination throws exception\nWhen marking EMR cluster for termination throws exception in version c7n-0.8.31.2. I see the tag created in cluster\r\n`````\r\npolicies:\r\n- name: emr-mark-clusters-for-termination\r\n  resource: emr\r\n  filters:\r\n    - type: value\r\n      key: \"Id\"\r\n      op: in\r\n      value:\r\n        - 'abcdefghij'\r\n  actions:\r\n    - type: mark-for-op\r\n      tag: 'custodian-emr-terminate'\r\n      op: terminate\r\n      days: 4\r\n`````\r\n\r\nthis policy throws exception \r\n\r\n2018-09-27 19:20:30,262: custodian.actions:INFO Tagging 1 resources for terminate on 2018/10/01\r\n2018-09-27 19:20:31,720: custodian.actions:ERROR Exception with tags: [{u'Value': u'Resource does not meet policy: terminate@2018/10/01', u'Key': 'custodian-emr-terminate'}] on resources: abcdefghij\r\n 'dict' object is not callable\r\n`\r\nThough the EMR is marked with tag ''custodian-emr-terminate', filtering on type: marked-for-op, returns 0 resources.\n", "before_files": [{"content": "# Copyright 2016-2017 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.\nfrom __future__ import absolute_import, division, print_function, unicode_literals\n\nimport logging\nimport time\n\nimport six\n\nfrom c7n.actions import ActionRegistry, BaseAction\nfrom c7n.exceptions import PolicyValidationError\nfrom c7n.filters import FilterRegistry, MetricsFilter\nfrom c7n.manager import resources\nfrom c7n.query import QueryResourceManager\nfrom c7n.utils import (\n    local_session, type_schema, get_retry)\nfrom c7n.tags import (\n    TagDelayedAction, RemoveTag, TagActionFilter, Tag)\n\nfilters = FilterRegistry('emr.filters')\nactions = ActionRegistry('emr.actions')\nlog = logging.getLogger('custodian.emr')\n\nfilters.register('marked-for-op', TagActionFilter)\n\n\n@resources.register('emr')\nclass EMRCluster(QueryResourceManager):\n    \"\"\"Resource manager for Elastic MapReduce clusters\n    \"\"\"\n\n    class resource_type(object):\n        service = 'emr'\n        type = 'emr'\n        cluster_states = ['WAITING', 'BOOTSTRAPPING', 'RUNNING', 'STARTING']\n        enum_spec = ('list_clusters', 'Clusters', {'ClusterStates': cluster_states})\n        name = 'Name'\n        id = 'Id'\n        date = \"Status.Timeline.CreationDateTime\"\n        filter_name = None\n        dimension = None\n\n    action_registry = actions\n    filter_registry = filters\n    retry = staticmethod(get_retry(('ThrottlingException',)))\n\n    def __init__(self, ctx, data):\n        super(EMRCluster, self).__init__(ctx, data)\n        self.queries = QueryFilter.parse(\n            self.data.get('query', [\n                {'ClusterStates': [\n                    'running', 'bootstrapping', 'waiting']}]))\n\n    @classmethod\n    def get_permissions(cls):\n        return (\"elasticmapreduce:ListClusters\",\n                \"elasticmapreduce:DescribeCluster\")\n\n    def get_resources(self, ids):\n        # no filtering by id set supported at the api\n        client = local_session(self.session_factory).client('emr')\n        results = []\n        for jid in ids:\n            results.append(\n                client.describe_cluster(ClusterId=jid)['Cluster'])\n        return results\n\n    def resources(self, query=None):\n        q = self.consolidate_query_filter()\n        if q is not None:\n            query = query or {}\n            for i in range(0, len(q)):\n                query[q[i]['Name']] = q[i]['Values']\n        return super(EMRCluster, self).resources(query=query)\n\n    def consolidate_query_filter(self):\n        result = []\n        names = set()\n        # allow same name to be specified multiple times and append the queries\n        # under the same name\n        for q in self.queries:\n            query_filter = q.query()\n            if query_filter['Name'] in names:\n                for filt in result:\n                    if query_filter['Name'] == filt['Name']:\n                        filt['Values'].extend(query_filter['Values'])\n            else:\n                names.add(query_filter['Name'])\n                result.append(query_filter)\n        if 'ClusterStates' not in names:\n            # include default query\n            result.append(\n                {\n                    'Name': 'ClusterStates',\n                    'Values': ['WAITING', 'RUNNING', 'BOOTSTRAPPING'],\n                }\n            )\n        return result\n\n    def augment(self, resources):\n        client = local_session(\n            self.get_resource_manager('emr').session_factory).client('emr')\n        result = []\n        # remap for cwmetrics\n        for r in resources:\n            cluster = self.retry(\n                client.describe_cluster, ClusterId=r['Id'])['Cluster']\n            result.append(cluster)\n        return result\n\n\n@EMRCluster.filter_registry.register('metrics')\nclass EMRMetrics(MetricsFilter):\n\n    def get_dimensions(self, resource):\n        # Job flow id is legacy name for cluster id\n        return [{'Name': 'JobFlowId', 'Value': resource['Id']}]\n\n\n@actions.register('mark-for-op')\nclass TagDelayedAction(TagDelayedAction):\n    \"\"\"Action to specify an action to occur at a later date\n\n    :example:\n\n    .. code-block:: yaml\n\n            policies:\n              - name: emr-mark-for-op\n                resource: emr\n                filters:\n                  - \"tag:Name\": absent\n                actions:\n                  - type: mark-for-op\n                    tag: custodian_cleanup\n                    op: terminate\n                    days: 4\n                    msg: \"Cluster does not have required tags\"\n    \"\"\"\n\n    permission = ('elasticmapreduce:AddTags',)\n    batch_size = 1\n    retry = staticmethod(get_retry(('ThrottlingException',)))\n\n    def process_resource_set(self, resources, tags):\n        client = local_session(\n            self.manager.session_factory).client('emr')\n        for r in resources:\n            self.retry(client.add_tags(ResourceId=r['Id'], Tags=tags))\n\n\n@actions.register('tag')\nclass TagTable(Tag):\n    \"\"\"Action to create tag(s) on a resource\n\n    :example:\n\n    .. code-block:: yaml\n\n            policies:\n              - name: emr-tag-table\n                resource: emr\n                filters:\n                  - \"tag:target-tag\": absent\n                actions:\n                  - type: tag\n                    key: target-tag\n                    value: target-tag-value\n    \"\"\"\n\n    permissions = ('elasticmapreduce:AddTags',)\n    batch_size = 1\n    retry = staticmethod(get_retry(('ThrottlingException',)))\n\n    def process_resource_set(self, resources, tags):\n        client = local_session(self.manager.session_factory).client('emr')\n        for r in resources:\n            self.retry(client.add_tags(ResourceId=r['Id'], Tags=tags))\n\n\n@actions.register('remove-tag')\nclass UntagTable(RemoveTag):\n    \"\"\"Action to remove tag(s) on a resource\n\n    :example:\n\n    .. code-block:: yaml\n\n            policies:\n              - name: emr-remove-tag\n                resource: emr\n                filters:\n                  - \"tag:target-tag\": present\n                actions:\n                  - type: remove-tag\n                    tags: [\"target-tag\"]\n    \"\"\"\n\n    concurrency = 2\n    batch_size = 5\n    permissions = ('elasticmapreduce:RemoveTags',)\n\n    def process_resource_set(self, resources, tag_keys):\n        client = local_session(\n            self.manager.session_factory).client('emr')\n        for r in resources:\n            client.remove_tags(\n                ResourceId=r['Id'], TagKeys=tag_keys)\n\n\n@actions.register('terminate')\nclass Terminate(BaseAction):\n    \"\"\"Action to terminate EMR cluster(s)\n\n    It is recommended to apply a filter to the terminate action to avoid\n    termination of all EMR clusters\n\n    :example:\n\n    .. code-block:: yaml\n\n            policies:\n              - name: emr-terminate\n                resource: emr\n                query:\n                  - ClusterStates: [STARTING, BOOTSTRAPPING, RUNNING, WAITING]\n                actions:\n                  - terminate\n    \"\"\"\n\n    schema = type_schema('terminate', force={'type': 'boolean'})\n    permissions = (\"elasticmapreduce:TerminateJobFlows\",)\n    delay = 5\n\n    def process(self, emrs):\n        client = local_session(self.manager.session_factory).client('emr')\n        cluster_ids = [emr['Id'] for emr in emrs]\n        if self.data.get('force'):\n            client.set_termination_protection(\n                JobFlowIds=cluster_ids, TerminationProtected=False)\n            time.sleep(self.delay)\n        client.terminate_job_flows(JobFlowIds=cluster_ids)\n        self.log.info(\"Deleted emrs: %s\", cluster_ids)\n        return emrs\n\n\n# Valid EMR Query Filters\nEMR_VALID_FILTERS = set(('CreatedAfter', 'CreatedBefore', 'ClusterStates'))\n\n\nclass QueryFilter(object):\n\n    @classmethod\n    def parse(cls, data):\n        results = []\n        for d in data:\n            if not isinstance(d, dict):\n                raise PolicyValidationError(\n                    \"EMR Query Filter Invalid structure %s\" % d)\n            results.append(cls(d).validate())\n        return results\n\n    def __init__(self, data):\n        self.data = data\n        self.key = None\n        self.value = None\n\n    def validate(self):\n        if not len(list(self.data.keys())) == 1:\n            raise PolicyValidationError(\n                \"EMR Query Filter Invalid %s\" % self.data)\n        self.key = list(self.data.keys())[0]\n        self.value = list(self.data.values())[0]\n\n        if self.key not in EMR_VALID_FILTERS and not self.key.startswith(\n                'tag:'):\n            raise PolicyValidationError(\n                \"EMR Query Filter invalid filter name %s\" % (self.data))\n\n        if self.value is None:\n            raise PolicyValidationError(\n                \"EMR Query Filters must have a value, use tag-key\"\n                \" w/ tag name as value for tag present checks\"\n                \" %s\" % self.data)\n        return self\n\n    def query(self):\n        value = self.value\n        if isinstance(self.value, six.string_types):\n            value = [self.value]\n\n        return {'Name': self.key, 'Values': value}\n", "path": "c7n/resources/emr.py"}], "after_files": [{"content": "# Copyright 2016-2017 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.\nfrom __future__ import absolute_import, division, print_function, unicode_literals\n\nimport logging\nimport time\n\nimport six\n\nfrom c7n.actions import ActionRegistry, BaseAction\nfrom c7n.exceptions import PolicyValidationError\nfrom c7n.filters import FilterRegistry, MetricsFilter\nfrom c7n.manager import resources\nfrom c7n.query import QueryResourceManager\nfrom c7n.utils import (\n    local_session, type_schema, get_retry)\nfrom c7n.tags import (\n    TagDelayedAction, RemoveTag, TagActionFilter, Tag)\n\nfilters = FilterRegistry('emr.filters')\nactions = ActionRegistry('emr.actions')\nlog = logging.getLogger('custodian.emr')\n\nfilters.register('marked-for-op', TagActionFilter)\n\n\n@resources.register('emr')\nclass EMRCluster(QueryResourceManager):\n    \"\"\"Resource manager for Elastic MapReduce clusters\n    \"\"\"\n\n    class resource_type(object):\n        service = 'emr'\n        type = 'emr'\n        cluster_states = ['WAITING', 'BOOTSTRAPPING', 'RUNNING', 'STARTING']\n        enum_spec = ('list_clusters', 'Clusters', {'ClusterStates': cluster_states})\n        name = 'Name'\n        id = 'Id'\n        date = \"Status.Timeline.CreationDateTime\"\n        filter_name = None\n        dimension = None\n\n    action_registry = actions\n    filter_registry = filters\n    retry = staticmethod(get_retry(('ThrottlingException',)))\n\n    def __init__(self, ctx, data):\n        super(EMRCluster, self).__init__(ctx, data)\n        self.queries = QueryFilter.parse(\n            self.data.get('query', [\n                {'ClusterStates': [\n                    'running', 'bootstrapping', 'waiting']}]))\n\n    @classmethod\n    def get_permissions(cls):\n        return (\"elasticmapreduce:ListClusters\",\n                \"elasticmapreduce:DescribeCluster\")\n\n    def get_resources(self, ids):\n        # no filtering by id set supported at the api\n        client = local_session(self.session_factory).client('emr')\n        results = []\n        for jid in ids:\n            results.append(\n                client.describe_cluster(ClusterId=jid)['Cluster'])\n        return results\n\n    def resources(self, query=None):\n        q = self.consolidate_query_filter()\n        if q is not None:\n            query = query or {}\n            for i in range(0, len(q)):\n                query[q[i]['Name']] = q[i]['Values']\n        return super(EMRCluster, self).resources(query=query)\n\n    def consolidate_query_filter(self):\n        result = []\n        names = set()\n        # allow same name to be specified multiple times and append the queries\n        # under the same name\n        for q in self.queries:\n            query_filter = q.query()\n            if query_filter['Name'] in names:\n                for filt in result:\n                    if query_filter['Name'] == filt['Name']:\n                        filt['Values'].extend(query_filter['Values'])\n            else:\n                names.add(query_filter['Name'])\n                result.append(query_filter)\n        if 'ClusterStates' not in names:\n            # include default query\n            result.append(\n                {\n                    'Name': 'ClusterStates',\n                    'Values': ['WAITING', 'RUNNING', 'BOOTSTRAPPING'],\n                }\n            )\n        return result\n\n    def augment(self, resources):\n        client = local_session(\n            self.get_resource_manager('emr').session_factory).client('emr')\n        result = []\n        # remap for cwmetrics\n        for r in resources:\n            cluster = self.retry(\n                client.describe_cluster, ClusterId=r['Id'])['Cluster']\n            result.append(cluster)\n        return result\n\n\n@EMRCluster.filter_registry.register('metrics')\nclass EMRMetrics(MetricsFilter):\n\n    def get_dimensions(self, resource):\n        # Job flow id is legacy name for cluster id\n        return [{'Name': 'JobFlowId', 'Value': resource['Id']}]\n\n\n@actions.register('mark-for-op')\nclass TagDelayedAction(TagDelayedAction):\n    \"\"\"Action to specify an action to occur at a later date\n\n    :example:\n\n    .. code-block:: yaml\n\n            policies:\n              - name: emr-mark-for-op\n                resource: emr\n                filters:\n                  - \"tag:Name\": absent\n                actions:\n                  - type: mark-for-op\n                    tag: custodian_cleanup\n                    op: terminate\n                    days: 4\n                    msg: \"Cluster does not have required tags\"\n    \"\"\"\n\n    permission = ('elasticmapreduce:AddTags',)\n    batch_size = 1\n    retry = staticmethod(get_retry(('ThrottlingException',)))\n\n    def process_resource_set(self, resources, tags):\n        client = local_session(\n            self.manager.session_factory).client('emr')\n        for r in resources:\n            self.retry(client.add_tags, ResourceId=r['Id'], Tags=tags)\n\n\n@actions.register('tag')\nclass TagTable(Tag):\n    \"\"\"Action to create tag(s) on a resource\n\n    :example:\n\n    .. code-block:: yaml\n\n            policies:\n              - name: emr-tag-table\n                resource: emr\n                filters:\n                  - \"tag:target-tag\": absent\n                actions:\n                  - type: tag\n                    key: target-tag\n                    value: target-tag-value\n    \"\"\"\n\n    permissions = ('elasticmapreduce:AddTags',)\n    batch_size = 1\n    retry = staticmethod(get_retry(('ThrottlingException',)))\n\n    def process_resource_set(self, resources, tags):\n        client = local_session(self.manager.session_factory).client('emr')\n        for r in resources:\n            self.retry(client.add_tags, ResourceId=r['Id'], Tags=tags)\n\n\n@actions.register('remove-tag')\nclass UntagTable(RemoveTag):\n    \"\"\"Action to remove tag(s) on a resource\n\n    :example:\n\n    .. code-block:: yaml\n\n            policies:\n              - name: emr-remove-tag\n                resource: emr\n                filters:\n                  - \"tag:target-tag\": present\n                actions:\n                  - type: remove-tag\n                    tags: [\"target-tag\"]\n    \"\"\"\n\n    concurrency = 2\n    batch_size = 5\n    permissions = ('elasticmapreduce:RemoveTags',)\n\n    def process_resource_set(self, resources, tag_keys):\n        client = local_session(\n            self.manager.session_factory).client('emr')\n        for r in resources:\n            client.remove_tags(\n                ResourceId=r['Id'], TagKeys=tag_keys)\n\n\n@actions.register('terminate')\nclass Terminate(BaseAction):\n    \"\"\"Action to terminate EMR cluster(s)\n\n    It is recommended to apply a filter to the terminate action to avoid\n    termination of all EMR clusters\n\n    :example:\n\n    .. code-block:: yaml\n\n            policies:\n              - name: emr-terminate\n                resource: emr\n                query:\n                  - ClusterStates: [STARTING, BOOTSTRAPPING, RUNNING, WAITING]\n                actions:\n                  - terminate\n    \"\"\"\n\n    schema = type_schema('terminate', force={'type': 'boolean'})\n    permissions = (\"elasticmapreduce:TerminateJobFlows\",)\n    delay = 5\n\n    def process(self, emrs):\n        client = local_session(self.manager.session_factory).client('emr')\n        cluster_ids = [emr['Id'] for emr in emrs]\n        if self.data.get('force'):\n            client.set_termination_protection(\n                JobFlowIds=cluster_ids, TerminationProtected=False)\n            time.sleep(self.delay)\n        client.terminate_job_flows(JobFlowIds=cluster_ids)\n        self.log.info(\"Deleted emrs: %s\", cluster_ids)\n        return emrs\n\n\n# Valid EMR Query Filters\nEMR_VALID_FILTERS = set(('CreatedAfter', 'CreatedBefore', 'ClusterStates'))\n\n\nclass QueryFilter(object):\n\n    @classmethod\n    def parse(cls, data):\n        results = []\n        for d in data:\n            if not isinstance(d, dict):\n                raise PolicyValidationError(\n                    \"EMR Query Filter Invalid structure %s\" % d)\n            results.append(cls(d).validate())\n        return results\n\n    def __init__(self, data):\n        self.data = data\n        self.key = None\n        self.value = None\n\n    def validate(self):\n        if not len(list(self.data.keys())) == 1:\n            raise PolicyValidationError(\n                \"EMR Query Filter Invalid %s\" % self.data)\n        self.key = list(self.data.keys())[0]\n        self.value = list(self.data.values())[0]\n\n        if self.key not in EMR_VALID_FILTERS and not self.key.startswith(\n                'tag:'):\n            raise PolicyValidationError(\n                \"EMR Query Filter invalid filter name %s\" % (self.data))\n\n        if self.value is None:\n            raise PolicyValidationError(\n                \"EMR Query Filters must have a value, use tag-key\"\n                \" w/ tag name as value for tag present checks\"\n                \" %s\" % self.data)\n        return self\n\n    def query(self):\n        value = self.value\n        if isinstance(self.value, six.string_types):\n            value = [self.value]\n\n        return {'Name': self.key, 'Values': value}\n", "path": "c7n/resources/emr.py"}]}