We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
aws.elasticsearch Error Scanning More Than 5 domains
**Describe the bug**
When running any elasticsearch policy on an account region with more than 5 elasticsearch domains the policy now bombs out with the error - ```error:An error occurred (ValidationException) when calling the DescribeElasticsearchDomains operation: Please provide a maximum of 5 Elasticsearch domain names to describe.```

**To Reproduce**
Create 6 es domains and run an elasticsearch c7n policy, error will occur

**Expected behavior**
It should chunk the calls into domains of 5 or less


**Background (please complete the following information):**
 - OS: Ubuntu v20
 - Python Version: 3.8
 - Custodian Version: 0.9.2.0
 - Tool Version: [if applicable]
 - Cloud Provider: aws
 - Policy:  any policy which queries ES
 - Traceback: 
```
[ERROR]	2020-05-22T14:51:25.978Z	9ef7929b-b494-434e-9f9f-dfdfdfdfdfdfd Error while executing policy
Traceback (most recent call last):
  File "/var/task/c7n/policy.py", line 291, in run
    resources = self.policy.resource_manager.resources()
  File "/var/task/c7n/query.py", line 466, in resources
    resources = self.augment(resources)
  File "/var/task/c7n/query.py", line 521, in augment
    return self.source.augment(resources)
  File "/var/task/c7n/resources/elasticsearch.py", line 48, in augment
    return _augment(domains)
  File "/var/task/c7n/resources/elasticsearch.py", line 39, in _augment
    resources = self.manager.retry(
  File "/var/task/c7n/utils.py", line 373, in _retry
    return func(*args, **kw)
  File "/var/task/botocore/client.py", line 316, in _api_call
    return self._make_api_call(operation_name, kwargs)
  File "/var/task/botocore/client.py", line 635, in _make_api_call
    raise error_class(parsed_response, operation_name)
botocore.errorfactory.ValidationException: An error occurred (ValidationException) when calling the DescribeElasticsearchDomains operation: Please provide a maximum of 5 Elasticsearch domain names to describe.
```
 - `custodian version --debug` output

**Additional context**
Seems to be introduced with 0.9.2.0


--- 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/elasticsearch.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 jmespath
15 
16 from c7n.actions import Action, ModifyVpcSecurityGroupsAction
17 from c7n.filters import MetricsFilter
18 from c7n.filters.vpc import SecurityGroupFilter, SubnetFilter, VpcFilter
19 from c7n.manager import resources
20 from c7n.query import ConfigSource, DescribeSource, QueryResourceManager, TypeInfo
21 from c7n.utils import local_session, type_schema
22 from c7n.tags import Tag, RemoveTag, TagActionFilter, TagDelayedAction
23 
24 from .securityhub import PostFinding
25 
26 
27 class DescribeDomain(DescribeSource):
28 
29     def get_resources(self, resource_ids):
30         client = local_session(self.manager.session_factory).client('es')
31         return client.describe_elasticsearch_domains(
32             DomainNames=resource_ids)['DomainStatusList']
33 
34     def augment(self, domains):
35         client = local_session(self.manager.session_factory).client('es')
36         model = self.manager.get_model()
37 
38         def _augment(resource_set):
39             resources = self.manager.retry(
40                 client.describe_elasticsearch_domains,
41                 DomainNames=resource_set)['DomainStatusList']
42             for r in resources:
43                 rarn = self.manager.generate_arn(r[model.id])
44                 r['Tags'] = self.manager.retry(
45                     client.list_tags, ARN=rarn).get('TagList', [])
46             return resources
47 
48         return _augment(domains)
49 
50 
51 @resources.register('elasticsearch')
52 class ElasticSearchDomain(QueryResourceManager):
53 
54     class resource_type(TypeInfo):
55         service = 'es'
56         arn = 'ARN'
57         arn_type = 'domain'
58         enum_spec = (
59             'list_domain_names', 'DomainNames[].DomainName', None)
60         id = 'DomainName'
61         name = 'Name'
62         dimension = "DomainName"
63         cfn_type = config_type = 'AWS::Elasticsearch::Domain'
64 
65     source_mapping = {
66         'describe': DescribeDomain,
67         'config': ConfigSource
68     }
69 
70 
71 ElasticSearchDomain.filter_registry.register('marked-for-op', TagActionFilter)
72 
73 
74 @ElasticSearchDomain.filter_registry.register('subnet')
75 class Subnet(SubnetFilter):
76 
77     RelatedIdsExpression = "VPCOptions.SubnetIds[]"
78 
79 
80 @ElasticSearchDomain.filter_registry.register('security-group')
81 class SecurityGroup(SecurityGroupFilter):
82 
83     RelatedIdsExpression = "VPCOptions.SecurityGroupIds[]"
84 
85 
86 @ElasticSearchDomain.filter_registry.register('vpc')
87 class Vpc(VpcFilter):
88 
89     RelatedIdsExpression = "VPCOptions.VPCId"
90 
91 
92 @ElasticSearchDomain.filter_registry.register('metrics')
93 class Metrics(MetricsFilter):
94 
95     def get_dimensions(self, resource):
96         return [{'Name': 'ClientId',
97                  'Value': self.manager.account_id},
98                 {'Name': 'DomainName',
99                  'Value': resource['DomainName']}]
100 
101 
102 @ElasticSearchDomain.action_registry.register('post-finding')
103 class ElasticSearchPostFinding(PostFinding):
104 
105     resource_type = 'AwsElasticsearchDomain'
106 
107     def format_resource(self, r):
108         envelope, payload = self.format_envelope(r)
109         payload.update(self.filter_empty({
110             'AccessPolicies': r.get('AccessPolicies'),
111             'DomainId': r['DomainId'],
112             'DomainName': r['DomainName'],
113             'Endpoint': r.get('Endpoint'),
114             'Endpoints': r.get('Endpoints'),
115             'DomainEndpointOptions': self.filter_empty({
116                 'EnforceHTTPS': jmespath.search(
117                     'DomainEndpointOptions.EnforceHTTPS', r),
118                 'TLSSecurityPolicy': jmespath.search(
119                     'DomainEndpointOptions.TLSSecurityPolicy', r)
120             }),
121             'ElasticsearchVersion': r['ElasticsearchVersion'],
122             'EncryptionAtRestOptions': self.filter_empty({
123                 'Enabled': jmespath.search(
124                     'EncryptionAtRestOptions.Enabled', r),
125                 'KmsKeyId': jmespath.search(
126                     'EncryptionAtRestOptions.KmsKeyId', r)
127             }),
128             'NodeToNodeEncryptionOptions': self.filter_empty({
129                 'Enabled': jmespath.search(
130                     'NodeToNodeEncryptionOptions.Enabled', r)
131             }),
132             'VPCOptions': self.filter_empty({
133                 'AvailabilityZones': jmespath.search(
134                     'VPCOptions.AvailabilityZones', r),
135                 'SecurityGroupIds': jmespath.search(
136                     'VPCOptions.SecurityGroupIds', r),
137                 'SubnetIds': jmespath.search('VPCOptions.SubnetIds', r),
138                 'VPCId': jmespath.search('VPCOptions.VPCId', r)
139             })
140         }))
141         return envelope
142 
143 
144 @ElasticSearchDomain.action_registry.register('modify-security-groups')
145 class ElasticSearchModifySG(ModifyVpcSecurityGroupsAction):
146     """Modify security groups on an Elasticsearch domain"""
147 
148     permissions = ('es:UpdateElasticsearchDomainConfig',)
149 
150     def process(self, domains):
151         groups = super(ElasticSearchModifySG, self).get_groups(domains)
152         client = local_session(self.manager.session_factory).client('es')
153 
154         for dx, d in enumerate(domains):
155             client.update_elasticsearch_domain_config(
156                 DomainName=d['DomainName'],
157                 VPCOptions={
158                     'SecurityGroupIds': groups[dx]})
159 
160 
161 @ElasticSearchDomain.action_registry.register('delete')
162 class Delete(Action):
163 
164     schema = type_schema('delete')
165     permissions = ('es:DeleteElasticsearchDomain',)
166 
167     def process(self, resources):
168         client = local_session(self.manager.session_factory).client('es')
169         for r in resources:
170             client.delete_elasticsearch_domain(DomainName=r['DomainName'])
171 
172 
173 @ElasticSearchDomain.action_registry.register('tag')
174 class ElasticSearchAddTag(Tag):
175     """Action to create tag(s) on an existing elasticsearch domain
176 
177     :example:
178 
179     .. code-block:: yaml
180 
181                 policies:
182                   - name: es-add-tag
183                     resource: elasticsearch
184                     filters:
185                       - "tag:DesiredTag": absent
186                     actions:
187                       - type: tag
188                         key: DesiredTag
189                         value: DesiredValue
190     """
191     permissions = ('es:AddTags',)
192 
193     def process_resource_set(self, client, domains, tags):
194         for d in domains:
195             try:
196                 client.add_tags(ARN=d['ARN'], TagList=tags)
197             except client.exceptions.ResourceNotFoundExecption:
198                 continue
199 
200 
201 @ElasticSearchDomain.action_registry.register('remove-tag')
202 class ElasticSearchRemoveTag(RemoveTag):
203     """Removes tag(s) on an existing elasticsearch domain
204 
205     :example:
206 
207     .. code-block:: yaml
208 
209         policies:
210           - name: es-remove-tag
211             resource: elasticsearch
212             filters:
213               - "tag:ExpiredTag": present
214             actions:
215               - type: remove-tag
216                 tags: ['ExpiredTag']
217         """
218     permissions = ('es:RemoveTags',)
219 
220     def process_resource_set(self, client, domains, tags):
221         for d in domains:
222             try:
223                 client.remove_tags(ARN=d['ARN'], TagKeys=tags)
224             except client.exceptions.ResourceNotFoundExecption:
225                 continue
226 
227 
228 @ElasticSearchDomain.action_registry.register('mark-for-op')
229 class ElasticSearchMarkForOp(TagDelayedAction):
230     """Tag an elasticsearch domain for action later
231 
232     :example:
233 
234     .. code-block:: yaml
235 
236                 policies:
237                   - name: es-delete-missing
238                     resource: elasticsearch
239                     filters:
240                       - "tag:DesiredTag": absent
241                     actions:
242                       - type: mark-for-op
243                         days: 7
244                         op: delete
245                         tag: c7n_es_delete
246     """
247
```
--- 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/elasticsearch.py b/c7n/resources/elasticsearch.py
--- a/c7n/resources/elasticsearch.py
+++ b/c7n/resources/elasticsearch.py
@@ -18,7 +18,7 @@
 from c7n.filters.vpc import SecurityGroupFilter, SubnetFilter, VpcFilter
 from c7n.manager import resources
 from c7n.query import ConfigSource, DescribeSource, QueryResourceManager, TypeInfo
-from c7n.utils import local_session, type_schema
+from c7n.utils import chunks, local_session, type_schema
 from c7n.tags import Tag, RemoveTag, TagActionFilter, TagDelayedAction
 
 from .securityhub import PostFinding
@@ -34,6 +34,7 @@
     def augment(self, domains):
         client = local_session(self.manager.session_factory).client('es')
         model = self.manager.get_model()
+        results = []
 
         def _augment(resource_set):
             resources = self.manager.retry(
@@ -45,7 +46,10 @@
                     client.list_tags, ARN=rarn).get('TagList', [])
             return resources
 
-        return _augment(domains)
+        for resource_set in chunks(domains, 5):
+            results.extend(_augment(resource_set))
+
+        return results
 
 
 @resources.register('elasticsearch')

{"golden_diff": "diff --git a/c7n/resources/elasticsearch.py b/c7n/resources/elasticsearch.py\n--- a/c7n/resources/elasticsearch.py\n+++ b/c7n/resources/elasticsearch.py\n@@ -18,7 +18,7 @@\n from c7n.filters.vpc import SecurityGroupFilter, SubnetFilter, VpcFilter\n from c7n.manager import resources\n from c7n.query import ConfigSource, DescribeSource, QueryResourceManager, TypeInfo\n-from c7n.utils import local_session, type_schema\n+from c7n.utils import chunks, local_session, type_schema\n from c7n.tags import Tag, RemoveTag, TagActionFilter, TagDelayedAction\n \n from .securityhub import PostFinding\n@@ -34,6 +34,7 @@\n     def augment(self, domains):\n         client = local_session(self.manager.session_factory).client('es')\n         model = self.manager.get_model()\n+        results = []\n \n         def _augment(resource_set):\n             resources = self.manager.retry(\n@@ -45,7 +46,10 @@\n                     client.list_tags, ARN=rarn).get('TagList', [])\n             return resources\n \n-        return _augment(domains)\n+        for resource_set in chunks(domains, 5):\n+            results.extend(_augment(resource_set))\n+\n+        return results\n \n \n @resources.register('elasticsearch')\n", "issue": "aws.elasticsearch Error Scanning More Than 5 domains\n**Describe the bug**\r\nWhen running any elasticsearch policy on an account region with more than 5 elasticsearch domains the policy now bombs out with the error - ```error:An error occurred (ValidationException) when calling the DescribeElasticsearchDomains operation: Please provide a maximum of 5 Elasticsearch domain names to describe.```\r\n\r\n**To Reproduce**\r\nCreate 6 es domains and run an elasticsearch c7n policy, error will occur\r\n\r\n**Expected behavior**\r\nIt should chunk the calls into domains of 5 or less\r\n\r\n\r\n**Background (please complete the following information):**\r\n - OS: Ubuntu v20\r\n - Python Version: 3.8\r\n - Custodian Version: 0.9.2.0\r\n - Tool Version: [if applicable]\r\n - Cloud Provider: aws\r\n - Policy:  any policy which queries ES\r\n - Traceback: \r\n```\r\n[ERROR]\t2020-05-22T14:51:25.978Z\t9ef7929b-b494-434e-9f9f-dfdfdfdfdfdfd Error while executing policy\r\nTraceback (most recent call last):\r\n  File \"/var/task/c7n/policy.py\", line 291, in run\r\n    resources = self.policy.resource_manager.resources()\r\n  File \"/var/task/c7n/query.py\", line 466, in resources\r\n    resources = self.augment(resources)\r\n  File \"/var/task/c7n/query.py\", line 521, in augment\r\n    return self.source.augment(resources)\r\n  File \"/var/task/c7n/resources/elasticsearch.py\", line 48, in augment\r\n    return _augment(domains)\r\n  File \"/var/task/c7n/resources/elasticsearch.py\", line 39, in _augment\r\n    resources = self.manager.retry(\r\n  File \"/var/task/c7n/utils.py\", line 373, in _retry\r\n    return func(*args, **kw)\r\n  File \"/var/task/botocore/client.py\", line 316, in _api_call\r\n    return self._make_api_call(operation_name, kwargs)\r\n  File \"/var/task/botocore/client.py\", line 635, in _make_api_call\r\n    raise error_class(parsed_response, operation_name)\r\nbotocore.errorfactory.ValidationException: An error occurred (ValidationException) when calling the DescribeElasticsearchDomains operation: Please provide a maximum of 5 Elasticsearch domain names to describe.\r\n```\r\n - `custodian version --debug` output\r\n\r\n**Additional context**\r\nSeems to be introduced with 0.9.2.0\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 jmespath\n\nfrom c7n.actions import Action, ModifyVpcSecurityGroupsAction\nfrom c7n.filters import MetricsFilter\nfrom c7n.filters.vpc import SecurityGroupFilter, SubnetFilter, VpcFilter\nfrom c7n.manager import resources\nfrom c7n.query import ConfigSource, DescribeSource, QueryResourceManager, TypeInfo\nfrom c7n.utils import local_session, type_schema\nfrom c7n.tags import Tag, RemoveTag, TagActionFilter, TagDelayedAction\n\nfrom .securityhub import PostFinding\n\n\nclass DescribeDomain(DescribeSource):\n\n    def get_resources(self, resource_ids):\n        client = local_session(self.manager.session_factory).client('es')\n        return client.describe_elasticsearch_domains(\n            DomainNames=resource_ids)['DomainStatusList']\n\n    def augment(self, domains):\n        client = local_session(self.manager.session_factory).client('es')\n        model = self.manager.get_model()\n\n        def _augment(resource_set):\n            resources = self.manager.retry(\n                client.describe_elasticsearch_domains,\n                DomainNames=resource_set)['DomainStatusList']\n            for r in resources:\n                rarn = self.manager.generate_arn(r[model.id])\n                r['Tags'] = self.manager.retry(\n                    client.list_tags, ARN=rarn).get('TagList', [])\n            return resources\n\n        return _augment(domains)\n\n\n@resources.register('elasticsearch')\nclass ElasticSearchDomain(QueryResourceManager):\n\n    class resource_type(TypeInfo):\n        service = 'es'\n        arn = 'ARN'\n        arn_type = 'domain'\n        enum_spec = (\n            'list_domain_names', 'DomainNames[].DomainName', None)\n        id = 'DomainName'\n        name = 'Name'\n        dimension = \"DomainName\"\n        cfn_type = config_type = 'AWS::Elasticsearch::Domain'\n\n    source_mapping = {\n        'describe': DescribeDomain,\n        'config': ConfigSource\n    }\n\n\nElasticSearchDomain.filter_registry.register('marked-for-op', TagActionFilter)\n\n\n@ElasticSearchDomain.filter_registry.register('subnet')\nclass Subnet(SubnetFilter):\n\n    RelatedIdsExpression = \"VPCOptions.SubnetIds[]\"\n\n\n@ElasticSearchDomain.filter_registry.register('security-group')\nclass SecurityGroup(SecurityGroupFilter):\n\n    RelatedIdsExpression = \"VPCOptions.SecurityGroupIds[]\"\n\n\n@ElasticSearchDomain.filter_registry.register('vpc')\nclass Vpc(VpcFilter):\n\n    RelatedIdsExpression = \"VPCOptions.VPCId\"\n\n\n@ElasticSearchDomain.filter_registry.register('metrics')\nclass Metrics(MetricsFilter):\n\n    def get_dimensions(self, resource):\n        return [{'Name': 'ClientId',\n                 'Value': self.manager.account_id},\n                {'Name': 'DomainName',\n                 'Value': resource['DomainName']}]\n\n\n@ElasticSearchDomain.action_registry.register('post-finding')\nclass ElasticSearchPostFinding(PostFinding):\n\n    resource_type = 'AwsElasticsearchDomain'\n\n    def format_resource(self, r):\n        envelope, payload = self.format_envelope(r)\n        payload.update(self.filter_empty({\n            'AccessPolicies': r.get('AccessPolicies'),\n            'DomainId': r['DomainId'],\n            'DomainName': r['DomainName'],\n            'Endpoint': r.get('Endpoint'),\n            'Endpoints': r.get('Endpoints'),\n            'DomainEndpointOptions': self.filter_empty({\n                'EnforceHTTPS': jmespath.search(\n                    'DomainEndpointOptions.EnforceHTTPS', r),\n                'TLSSecurityPolicy': jmespath.search(\n                    'DomainEndpointOptions.TLSSecurityPolicy', r)\n            }),\n            'ElasticsearchVersion': r['ElasticsearchVersion'],\n            'EncryptionAtRestOptions': self.filter_empty({\n                'Enabled': jmespath.search(\n                    'EncryptionAtRestOptions.Enabled', r),\n                'KmsKeyId': jmespath.search(\n                    'EncryptionAtRestOptions.KmsKeyId', r)\n            }),\n            'NodeToNodeEncryptionOptions': self.filter_empty({\n                'Enabled': jmespath.search(\n                    'NodeToNodeEncryptionOptions.Enabled', r)\n            }),\n            'VPCOptions': self.filter_empty({\n                'AvailabilityZones': jmespath.search(\n                    'VPCOptions.AvailabilityZones', r),\n                'SecurityGroupIds': jmespath.search(\n                    'VPCOptions.SecurityGroupIds', r),\n                'SubnetIds': jmespath.search('VPCOptions.SubnetIds', r),\n                'VPCId': jmespath.search('VPCOptions.VPCId', r)\n            })\n        }))\n        return envelope\n\n\n@ElasticSearchDomain.action_registry.register('modify-security-groups')\nclass ElasticSearchModifySG(ModifyVpcSecurityGroupsAction):\n    \"\"\"Modify security groups on an Elasticsearch domain\"\"\"\n\n    permissions = ('es:UpdateElasticsearchDomainConfig',)\n\n    def process(self, domains):\n        groups = super(ElasticSearchModifySG, self).get_groups(domains)\n        client = local_session(self.manager.session_factory).client('es')\n\n        for dx, d in enumerate(domains):\n            client.update_elasticsearch_domain_config(\n                DomainName=d['DomainName'],\n                VPCOptions={\n                    'SecurityGroupIds': groups[dx]})\n\n\n@ElasticSearchDomain.action_registry.register('delete')\nclass Delete(Action):\n\n    schema = type_schema('delete')\n    permissions = ('es:DeleteElasticsearchDomain',)\n\n    def process(self, resources):\n        client = local_session(self.manager.session_factory).client('es')\n        for r in resources:\n            client.delete_elasticsearch_domain(DomainName=r['DomainName'])\n\n\n@ElasticSearchDomain.action_registry.register('tag')\nclass ElasticSearchAddTag(Tag):\n    \"\"\"Action to create tag(s) on an existing elasticsearch domain\n\n    :example:\n\n    .. code-block:: yaml\n\n                policies:\n                  - name: es-add-tag\n                    resource: elasticsearch\n                    filters:\n                      - \"tag:DesiredTag\": absent\n                    actions:\n                      - type: tag\n                        key: DesiredTag\n                        value: DesiredValue\n    \"\"\"\n    permissions = ('es:AddTags',)\n\n    def process_resource_set(self, client, domains, tags):\n        for d in domains:\n            try:\n                client.add_tags(ARN=d['ARN'], TagList=tags)\n            except client.exceptions.ResourceNotFoundExecption:\n                continue\n\n\n@ElasticSearchDomain.action_registry.register('remove-tag')\nclass ElasticSearchRemoveTag(RemoveTag):\n    \"\"\"Removes tag(s) on an existing elasticsearch domain\n\n    :example:\n\n    .. code-block:: yaml\n\n        policies:\n          - name: es-remove-tag\n            resource: elasticsearch\n            filters:\n              - \"tag:ExpiredTag\": present\n            actions:\n              - type: remove-tag\n                tags: ['ExpiredTag']\n        \"\"\"\n    permissions = ('es:RemoveTags',)\n\n    def process_resource_set(self, client, domains, tags):\n        for d in domains:\n            try:\n                client.remove_tags(ARN=d['ARN'], TagKeys=tags)\n            except client.exceptions.ResourceNotFoundExecption:\n                continue\n\n\n@ElasticSearchDomain.action_registry.register('mark-for-op')\nclass ElasticSearchMarkForOp(TagDelayedAction):\n    \"\"\"Tag an elasticsearch domain for action later\n\n    :example:\n\n    .. code-block:: yaml\n\n                policies:\n                  - name: es-delete-missing\n                    resource: elasticsearch\n                    filters:\n                      - \"tag:DesiredTag\": absent\n                    actions:\n                      - type: mark-for-op\n                        days: 7\n                        op: delete\n                        tag: c7n_es_delete\n    \"\"\"\n", "path": "c7n/resources/elasticsearch.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 jmespath\n\nfrom c7n.actions import Action, ModifyVpcSecurityGroupsAction\nfrom c7n.filters import MetricsFilter\nfrom c7n.filters.vpc import SecurityGroupFilter, SubnetFilter, VpcFilter\nfrom c7n.manager import resources\nfrom c7n.query import ConfigSource, DescribeSource, QueryResourceManager, TypeInfo\nfrom c7n.utils import chunks, local_session, type_schema\nfrom c7n.tags import Tag, RemoveTag, TagActionFilter, TagDelayedAction\n\nfrom .securityhub import PostFinding\n\n\nclass DescribeDomain(DescribeSource):\n\n    def get_resources(self, resource_ids):\n        client = local_session(self.manager.session_factory).client('es')\n        return client.describe_elasticsearch_domains(\n            DomainNames=resource_ids)['DomainStatusList']\n\n    def augment(self, domains):\n        client = local_session(self.manager.session_factory).client('es')\n        model = self.manager.get_model()\n        results = []\n\n        def _augment(resource_set):\n            resources = self.manager.retry(\n                client.describe_elasticsearch_domains,\n                DomainNames=resource_set)['DomainStatusList']\n            for r in resources:\n                rarn = self.manager.generate_arn(r[model.id])\n                r['Tags'] = self.manager.retry(\n                    client.list_tags, ARN=rarn).get('TagList', [])\n            return resources\n\n        for resource_set in chunks(domains, 5):\n            results.extend(_augment(resource_set))\n\n        return results\n\n\n@resources.register('elasticsearch')\nclass ElasticSearchDomain(QueryResourceManager):\n\n    class resource_type(TypeInfo):\n        service = 'es'\n        arn = 'ARN'\n        arn_type = 'domain'\n        enum_spec = (\n            'list_domain_names', 'DomainNames[].DomainName', None)\n        id = 'DomainName'\n        name = 'Name'\n        dimension = \"DomainName\"\n        cfn_type = config_type = 'AWS::Elasticsearch::Domain'\n\n    source_mapping = {\n        'describe': DescribeDomain,\n        'config': ConfigSource\n    }\n\n\nElasticSearchDomain.filter_registry.register('marked-for-op', TagActionFilter)\n\n\n@ElasticSearchDomain.filter_registry.register('subnet')\nclass Subnet(SubnetFilter):\n\n    RelatedIdsExpression = \"VPCOptions.SubnetIds[]\"\n\n\n@ElasticSearchDomain.filter_registry.register('security-group')\nclass SecurityGroup(SecurityGroupFilter):\n\n    RelatedIdsExpression = \"VPCOptions.SecurityGroupIds[]\"\n\n\n@ElasticSearchDomain.filter_registry.register('vpc')\nclass Vpc(VpcFilter):\n\n    RelatedIdsExpression = \"VPCOptions.VPCId\"\n\n\n@ElasticSearchDomain.filter_registry.register('metrics')\nclass Metrics(MetricsFilter):\n\n    def get_dimensions(self, resource):\n        return [{'Name': 'ClientId',\n                 'Value': self.manager.account_id},\n                {'Name': 'DomainName',\n                 'Value': resource['DomainName']}]\n\n\n@ElasticSearchDomain.action_registry.register('post-finding')\nclass ElasticSearchPostFinding(PostFinding):\n\n    resource_type = 'AwsElasticsearchDomain'\n\n    def format_resource(self, r):\n        envelope, payload = self.format_envelope(r)\n        payload.update(self.filter_empty({\n            'AccessPolicies': r.get('AccessPolicies'),\n            'DomainId': r['DomainId'],\n            'DomainName': r['DomainName'],\n            'Endpoint': r.get('Endpoint'),\n            'Endpoints': r.get('Endpoints'),\n            'DomainEndpointOptions': self.filter_empty({\n                'EnforceHTTPS': jmespath.search(\n                    'DomainEndpointOptions.EnforceHTTPS', r),\n                'TLSSecurityPolicy': jmespath.search(\n                    'DomainEndpointOptions.TLSSecurityPolicy', r)\n            }),\n            'ElasticsearchVersion': r['ElasticsearchVersion'],\n            'EncryptionAtRestOptions': self.filter_empty({\n                'Enabled': jmespath.search(\n                    'EncryptionAtRestOptions.Enabled', r),\n                'KmsKeyId': jmespath.search(\n                    'EncryptionAtRestOptions.KmsKeyId', r)\n            }),\n            'NodeToNodeEncryptionOptions': self.filter_empty({\n                'Enabled': jmespath.search(\n                    'NodeToNodeEncryptionOptions.Enabled', r)\n            }),\n            'VPCOptions': self.filter_empty({\n                'AvailabilityZones': jmespath.search(\n                    'VPCOptions.AvailabilityZones', r),\n                'SecurityGroupIds': jmespath.search(\n                    'VPCOptions.SecurityGroupIds', r),\n                'SubnetIds': jmespath.search('VPCOptions.SubnetIds', r),\n                'VPCId': jmespath.search('VPCOptions.VPCId', r)\n            })\n        }))\n        return envelope\n\n\n@ElasticSearchDomain.action_registry.register('modify-security-groups')\nclass ElasticSearchModifySG(ModifyVpcSecurityGroupsAction):\n    \"\"\"Modify security groups on an Elasticsearch domain\"\"\"\n\n    permissions = ('es:UpdateElasticsearchDomainConfig',)\n\n    def process(self, domains):\n        groups = super(ElasticSearchModifySG, self).get_groups(domains)\n        client = local_session(self.manager.session_factory).client('es')\n\n        for dx, d in enumerate(domains):\n            client.update_elasticsearch_domain_config(\n                DomainName=d['DomainName'],\n                VPCOptions={\n                    'SecurityGroupIds': groups[dx]})\n\n\n@ElasticSearchDomain.action_registry.register('delete')\nclass Delete(Action):\n\n    schema = type_schema('delete')\n    permissions = ('es:DeleteElasticsearchDomain',)\n\n    def process(self, resources):\n        client = local_session(self.manager.session_factory).client('es')\n        for r in resources:\n            client.delete_elasticsearch_domain(DomainName=r['DomainName'])\n\n\n@ElasticSearchDomain.action_registry.register('tag')\nclass ElasticSearchAddTag(Tag):\n    \"\"\"Action to create tag(s) on an existing elasticsearch domain\n\n    :example:\n\n    .. code-block:: yaml\n\n                policies:\n                  - name: es-add-tag\n                    resource: elasticsearch\n                    filters:\n                      - \"tag:DesiredTag\": absent\n                    actions:\n                      - type: tag\n                        key: DesiredTag\n                        value: DesiredValue\n    \"\"\"\n    permissions = ('es:AddTags',)\n\n    def process_resource_set(self, client, domains, tags):\n        for d in domains:\n            try:\n                client.add_tags(ARN=d['ARN'], TagList=tags)\n            except client.exceptions.ResourceNotFoundExecption:\n                continue\n\n\n@ElasticSearchDomain.action_registry.register('remove-tag')\nclass ElasticSearchRemoveTag(RemoveTag):\n    \"\"\"Removes tag(s) on an existing elasticsearch domain\n\n    :example:\n\n    .. code-block:: yaml\n\n        policies:\n          - name: es-remove-tag\n            resource: elasticsearch\n            filters:\n              - \"tag:ExpiredTag\": present\n            actions:\n              - type: remove-tag\n                tags: ['ExpiredTag']\n        \"\"\"\n    permissions = ('es:RemoveTags',)\n\n    def process_resource_set(self, client, domains, tags):\n        for d in domains:\n            try:\n                client.remove_tags(ARN=d['ARN'], TagKeys=tags)\n            except client.exceptions.ResourceNotFoundExecption:\n                continue\n\n\n@ElasticSearchDomain.action_registry.register('mark-for-op')\nclass ElasticSearchMarkForOp(TagDelayedAction):\n    \"\"\"Tag an elasticsearch domain for action later\n\n    :example:\n\n    .. code-block:: yaml\n\n                policies:\n                  - name: es-delete-missing\n                    resource: elasticsearch\n                    filters:\n                      - \"tag:DesiredTag\": absent\n                    actions:\n                      - type: mark-for-op\n                        days: 7\n                        op: delete\n                        tag: c7n_es_delete\n    \"\"\"\n", "path": "c7n/resources/elasticsearch.py"}]}