We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
emr extend network-location feature
**Is your feature request related to a problem? Please describe.**
Require subnet/sg tag comparison for resource: emr

**Describe the solution you'd like**
Extend filter `subnet`, `security-group` and `network-location` for emr

**Describe alternatives you've considered**
A clear and concise description of any alternative solutions or features you've considered.

**Additional context**
Add any other context or screenshots about the feature request here.


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

{"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@@ -14,6 +14,7 @@\n import logging\n import time\n import json\n+import jmespath\n \n from c7n.actions import ActionRegistry, BaseAction\n from c7n.exceptions import PolicyValidationError\n@@ -24,6 +25,7 @@\n     local_session, type_schema, get_retry)\n from c7n.tags import (\n     TagDelayedAction, RemoveTag, TagActionFilter, Tag)\n+import c7n.filters.vpc as net_filters\n \n filters = FilterRegistry('emr.filters')\n actions = ActionRegistry('emr.actions')\n@@ -286,6 +288,37 @@\n         return {'Name': self.key, 'Values': value}\n \n \n+@filters.register('subnet')\n+class SubnetFilter(net_filters.SubnetFilter):\n+\n+    RelatedIdsExpression = \"Ec2InstanceAttributes.RequestedEc2SubnetIds[]\"\n+\n+\n+@filters.register('security-group')\n+class SecurityGroupFilter(net_filters.SecurityGroupFilter):\n+\n+    RelatedIdsExpression = \"\"\n+    expressions = ('Ec2InstanceAttributes.EmrManagedMasterSecurityGroup',\n+                'Ec2InstanceAttributes.EmrManagedSlaveSecurityGroup',\n+                'Ec2InstanceAttributes.ServiceAccessSecurityGroup',\n+                'Ec2InstanceAttributes.AdditionalMasterSecurityGroups[]',\n+                'Ec2InstanceAttributes.AdditionalSlaveSecurityGroups[]')\n+\n+    def get_related_ids(self, resources):\n+        sg_ids = set()\n+        for r in resources:\n+            for exp in self.expressions:\n+                ids = jmespath.search(exp, r)\n+                if isinstance(ids, list):\n+                    sg_ids.update(tuple(ids))\n+                elif isinstance(ids, str):\n+                    sg_ids.add(ids)\n+        return list(sg_ids)\n+\n+\n+filters.register('network-location', net_filters.NetworkLocation)\n+\n+\n @resources.register('emr-security-configuration')\n class EMRSecurityConfiguration(QueryResourceManager):\n     \"\"\"Resource manager for EMR Security Configuration\n", "issue": "emr extend network-location feature\n**Is your feature request related to a problem? Please describe.**\r\nRequire subnet/sg tag comparison for resource: emr\r\n\r\n**Describe the solution you'd like**\r\nExtend filter `subnet`, `security-group` and `network-location` for emr\r\n\r\n**Describe alternatives you've considered**\r\nA clear and concise description of any alternative solutions or features you've considered.\r\n\r\n**Additional context**\r\nAdd any other context or screenshots about the feature request here.\r\n\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.\nimport logging\nimport time\nimport json\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, TypeInfo\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(TypeInfo):\n        service = 'emr'\n        arn_type = 'emr'\n        permission_prefix = 'elasticmapreduce'\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        cfn_type = 'AWS::EMR::Cluster'\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\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, client, resources, tags):\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, client, resources, tag_keys):\n        for r in resources:\n            client.remove_tags(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 = {'CreatedAfter', 'CreatedBefore', 'ClusterStates'}\n\n\nclass QueryFilter:\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, str):\n            value = [self.value]\n\n        return {'Name': self.key, 'Values': value}\n\n\n@resources.register('emr-security-configuration')\nclass EMRSecurityConfiguration(QueryResourceManager):\n    \"\"\"Resource manager for EMR Security Configuration\n    \"\"\"\n\n    class resource_type(TypeInfo):\n        service = 'emr'\n        arn_type = 'emr'\n        permission_prefix = 'elasticmapreduce'\n        enum_spec = ('list_security_configurations', 'SecurityConfigurations', None)\n        detail_spec = ('describe_security_configuration', 'Name', 'Name', None)\n        id = name = 'Name'\n        cfn_type = 'AWS::EMR::SecurityConfiguration'\n\n    permissions = ('elasticmapreduce:ListSecurityConfigurations',\n                  'elasticmapreduce:DescribeSecurityConfiguration',)\n\n    def augment(self, resources):\n        resources = super().augment(resources)\n        for r in resources:\n            r['SecurityConfiguration'] = json.loads(r['SecurityConfiguration'])\n        return resources\n\n\n@EMRSecurityConfiguration.action_registry.register('delete')\nclass DeleteEMRSecurityConfiguration(BaseAction):\n\n    schema = type_schema('delete')\n    permissions = ('elasticmapreduce:DeleteSecurityConfiguration',)\n\n    def process(self, resources):\n        client = local_session(self.manager.session_factory).client('emr')\n        for r in resources:\n            try:\n                client.delete_security_configuration(Name=r['Name'])\n            except client.exceptions.EntityNotFoundException:\n                continue\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.\nimport logging\nimport time\nimport json\nimport jmespath\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, TypeInfo\nfrom c7n.utils import (\n    local_session, type_schema, get_retry)\nfrom c7n.tags import (\n    TagDelayedAction, RemoveTag, TagActionFilter, Tag)\nimport c7n.filters.vpc as net_filters\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(TypeInfo):\n        service = 'emr'\n        arn_type = 'emr'\n        permission_prefix = 'elasticmapreduce'\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        cfn_type = 'AWS::EMR::Cluster'\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\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, client, resources, tags):\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, client, resources, tag_keys):\n        for r in resources:\n            client.remove_tags(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 = {'CreatedAfter', 'CreatedBefore', 'ClusterStates'}\n\n\nclass QueryFilter:\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, str):\n            value = [self.value]\n\n        return {'Name': self.key, 'Values': value}\n\n\n@filters.register('subnet')\nclass SubnetFilter(net_filters.SubnetFilter):\n\n    RelatedIdsExpression = \"Ec2InstanceAttributes.RequestedEc2SubnetIds[]\"\n\n\n@filters.register('security-group')\nclass SecurityGroupFilter(net_filters.SecurityGroupFilter):\n\n    RelatedIdsExpression = \"\"\n    expressions = ('Ec2InstanceAttributes.EmrManagedMasterSecurityGroup',\n                'Ec2InstanceAttributes.EmrManagedSlaveSecurityGroup',\n                'Ec2InstanceAttributes.ServiceAccessSecurityGroup',\n                'Ec2InstanceAttributes.AdditionalMasterSecurityGroups[]',\n                'Ec2InstanceAttributes.AdditionalSlaveSecurityGroups[]')\n\n    def get_related_ids(self, resources):\n        sg_ids = set()\n        for r in resources:\n            for exp in self.expressions:\n                ids = jmespath.search(exp, r)\n                if isinstance(ids, list):\n                    sg_ids.update(tuple(ids))\n                elif isinstance(ids, str):\n                    sg_ids.add(ids)\n        return list(sg_ids)\n\n\nfilters.register('network-location', net_filters.NetworkLocation)\n\n\n@resources.register('emr-security-configuration')\nclass EMRSecurityConfiguration(QueryResourceManager):\n    \"\"\"Resource manager for EMR Security Configuration\n    \"\"\"\n\n    class resource_type(TypeInfo):\n        service = 'emr'\n        arn_type = 'emr'\n        permission_prefix = 'elasticmapreduce'\n        enum_spec = ('list_security_configurations', 'SecurityConfigurations', None)\n        detail_spec = ('describe_security_configuration', 'Name', 'Name', None)\n        id = name = 'Name'\n        cfn_type = 'AWS::EMR::SecurityConfiguration'\n\n    permissions = ('elasticmapreduce:ListSecurityConfigurations',\n                  'elasticmapreduce:DescribeSecurityConfiguration',)\n\n    def augment(self, resources):\n        resources = super().augment(resources)\n        for r in resources:\n            r['SecurityConfiguration'] = json.loads(r['SecurityConfiguration'])\n        return resources\n\n\n@EMRSecurityConfiguration.action_registry.register('delete')\nclass DeleteEMRSecurityConfiguration(BaseAction):\n\n    schema = type_schema('delete')\n    permissions = ('elasticmapreduce:DeleteSecurityConfiguration',)\n\n    def process(self, resources):\n        client = local_session(self.manager.session_factory).client('emr')\n        for r in resources:\n            try:\n                client.delete_security_configuration(Name=r['Name'])\n            except client.exceptions.EntityNotFoundException:\n                continue\n", "path": "c7n/resources/emr.py"}]}