We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
S3Output should grant bucket owner full control
Currently `S3Output` misses the `ACL` option. In a cross account setup it is desirable to give the bucket owner full control.

Would you change the code like this?:
```
diff --git a/c7n/output.py b/c7n/output.py
index c3839c2f..5fb06f59 100644
--- a/c7n/output.py
+++ b/c7n/output.py
@@ -268,6 +268,7 @@ class S3Output(FSOutput):
                 self.transfer.upload_file(
                     os.path.join(root, f), self.bucket, key,
                     extra_args={
+                        'ACL': 'bucket-owner-full-control',
                         'ServerSideEncryption': 'AES256'})
 
     def use_s3(self):
```


--- 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/output.py`
Content:
```
1 # Copyright 2015-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 """
15 Outputs metrics, logs, structured records across
16 a variety of sources.
17 
18 See docs/usage/outputs.rst
19 
20 """
21 from __future__ import absolute_import, division, print_function, unicode_literals
22 
23 import datetime
24 import gzip
25 import logging
26 import shutil
27 import tempfile
28 
29 import os
30 
31 from boto3.s3.transfer import S3Transfer
32 
33 from c7n.registry import PluginRegistry
34 from c7n.log import CloudWatchLogHandler
35 from c7n.utils import local_session, parse_s3, get_retry
36 
37 DEFAULT_NAMESPACE = "CloudMaid"
38 
39 log = logging.getLogger('custodian.output')
40 
41 
42 blob_outputs = PluginRegistry('c7n.blob-outputs')
43 
44 
45 class MetricsOutput(object):
46     """Send metrics data to cloudwatch
47     """
48 
49     permissions = ("cloudWatch:PutMetricData",)
50 
51     retry = staticmethod(get_retry(('Throttling',)))
52 
53     @staticmethod
54     def select(metrics_enabled):
55         if metrics_enabled:
56             return MetricsOutput
57         return NullMetricsOutput
58 
59     def __init__(self, ctx, namespace=DEFAULT_NAMESPACE):
60         self.ctx = ctx
61         self.namespace = namespace
62         self.buf = []
63 
64     def get_timestamp(self):
65         """
66         Now, if C7N_METRICS_TZ is set to TRUE, UTC timestamp will be used.
67         For backwards compatibility, if it is not set, UTC will be the default.
68         To disable this and use the system's time zone, C7N_METRICS_TZ shoule be set to FALSE.
69         """
70 
71         if os.getenv("C7N_METRICS_TZ", '').upper() in ('TRUE', ''):
72             return datetime.datetime.utcnow()
73         else:
74             return datetime.datetime.now()
75 
76     def flush(self):
77         if self.buf:
78             self._put_metrics(self.namespace, self.buf)
79             self.buf = []
80 
81     def put_metric(self, key, value, unit, buffer=False, **dimensions):
82         d = {
83             "MetricName": key,
84             "Timestamp": self.get_timestamp(),
85             "Value": value,
86             "Unit": unit}
87         d["Dimensions"] = [
88             {"Name": "Policy", "Value": self.ctx.policy.name},
89             {"Name": "ResType", "Value": self.ctx.policy.resource_type}]
90         for k, v in dimensions.items():
91             d['Dimensions'].append({"Name": k, "Value": v})
92 
93         if buffer:
94             self.buf.append(d)
95             # Max metrics in a single request
96             if len(self.buf) == 20:
97                 self.flush()
98         else:
99             self._put_metrics(self.namespace, [d])
100 
101     def _put_metrics(self, ns, metrics):
102         watch = local_session(self.ctx.session_factory).client('cloudwatch')
103         return self.retry(
104             watch.put_metric_data, Namespace=ns, MetricData=metrics)
105 
106 
107 class NullMetricsOutput(MetricsOutput):
108 
109     permissions = ()
110 
111     def __init__(self, ctx, namespace=DEFAULT_NAMESPACE):
112         super(NullMetricsOutput, self).__init__(ctx, namespace)
113         self.data = []
114 
115     def _put_metrics(self, ns, metrics):
116         self.data.append({'Namespace': ns, 'MetricData': metrics})
117         for m in metrics:
118             if m['MetricName'] not in ('ActionTime', 'ResourceTime'):
119                 log.debug(self.format_metric(m))
120 
121     def format_metric(self, m):
122         label = "metric:%s %s:%s" % (m['MetricName'], m['Unit'], m['Value'])
123         for d in m['Dimensions']:
124             label += " %s:%s" % (d['Name'].lower(), d['Value'].lower())
125         return label
126 
127 
128 class LogOutput(object):
129 
130     log_format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
131 
132     def __init__(self, ctx):
133         self.ctx = ctx
134 
135     def get_handler(self):
136         raise NotImplementedError()
137 
138     def __enter__(self):
139         log.debug("Storing output with %s" % repr(self))
140         self.join_log()
141         return self
142 
143     def __exit__(self, exc_type=None, exc_value=None, exc_traceback=None):
144         self.leave_log()
145         if exc_type is not None:
146             log.exception("Error while executing policy")
147 
148     def join_log(self):
149         self.handler = self.get_handler()
150         self.handler.setLevel(logging.DEBUG)
151         self.handler.setFormatter(logging.Formatter(self.log_format))
152         mlog = logging.getLogger('custodian')
153         mlog.addHandler(self.handler)
154 
155     def leave_log(self):
156         mlog = logging.getLogger('custodian')
157         mlog.removeHandler(self.handler)
158         self.handler.flush()
159         self.handler.close()
160 
161 
162 class CloudWatchLogOutput(LogOutput):
163 
164     log_format = '%(asctime)s - %(levelname)s - %(name)s - %(message)s'
165 
166     def get_handler(self):
167         return CloudWatchLogHandler(
168             log_group=self.ctx.options.log_group,
169             log_stream=self.ctx.policy.name,
170             session_factory=lambda x=None: self.ctx.session_factory(
171                 assume=False))
172 
173     def __repr__(self):
174         return "<%s to group:%s stream:%s>" % (
175             self.__class__.__name__,
176             self.ctx.options.log_group,
177             self.ctx.policy.name)
178 
179 
180 class FSOutput(LogOutput):
181 
182     @staticmethod
183     def select(path):
184         for k in blob_outputs.keys():
185             if path.startswith('%s://' % k):
186                 return blob_outputs[k]
187         # Fall back local disk
188         return blob_outputs['file']
189 
190     @staticmethod
191     def join(*parts):
192         return os.path.join(*parts)
193 
194     def __init__(self, ctx):
195         super(FSOutput, self).__init__(ctx)
196         self.root_dir = self.ctx.output_path or tempfile.mkdtemp()
197 
198     def get_handler(self):
199         return logging.FileHandler(
200             os.path.join(self.root_dir, 'custodian-run.log'))
201 
202     def compress(self):
203         # Compress files individually so thats easy to walk them, without
204         # downloading tar and extracting.
205         for root, dirs, files in os.walk(self.root_dir):
206             for f in files:
207                 fp = os.path.join(root, f)
208                 with gzip.open(fp + ".gz", "wb", compresslevel=7) as zfh:
209                     with open(fp, "rb") as sfh:
210                         shutil.copyfileobj(sfh, zfh, length=2**15)
211                     os.remove(fp)
212 
213 
214 @blob_outputs.register('file')
215 class DirectoryOutput(FSOutput):
216 
217     permissions = ()
218 
219     def __init__(self, ctx):
220         super(DirectoryOutput, self).__init__(ctx)
221         if self.root_dir.startswith('file://'):
222             self.root_dir = self.root_dir[len('file://'):]
223         if self.ctx.output_path is not None:
224             if not os.path.exists(self.root_dir):
225                 os.makedirs(self.root_dir)
226 
227     def __repr__(self):
228         return "<%s to dir:%s>" % (self.__class__.__name__, self.root_dir)
229 
230 
231 @blob_outputs.register('s3')
232 class S3Output(FSOutput):
233     """
234     Usage:
235 
236     .. code-block:: python
237 
238        with S3Output(session_factory, 's3://bucket/prefix'):
239            log.info('xyz')  # -> log messages sent to custodian-run.log.gz
240 
241     """
242 
243     permissions = ('S3:PutObject',)
244 
245     def __init__(self, ctx):
246         super(S3Output, self).__init__(ctx)
247         self.date_path = datetime.datetime.now().strftime('%Y/%m/%d/%H')
248         self.s3_path, self.bucket, self.key_prefix = parse_s3(
249             self.ctx.output_path)
250         self.root_dir = tempfile.mkdtemp()
251         self.transfer = None
252 
253     def __repr__(self):
254         return "<%s to bucket:%s prefix:%s>" % (
255             self.__class__.__name__,
256             self.bucket,
257             "%s/%s" % (self.key_prefix, self.date_path))
258 
259     @staticmethod
260     def join(*parts):
261         return "/".join([s.strip('/') for s in parts])
262 
263     def __exit__(self, exc_type=None, exc_value=None, exc_traceback=None):
264         if exc_type is not None:
265             log.exception("Error while executing policy")
266         log.debug("Uploading policy logs")
267         self.leave_log()
268         self.compress()
269         self.transfer = S3Transfer(
270             self.ctx.session_factory(assume=False).client('s3'))
271         self.upload()
272         shutil.rmtree(self.root_dir)
273         log.debug("Policy Logs uploaded")
274 
275     def upload(self):
276         for root, dirs, files in os.walk(self.root_dir):
277             for f in files:
278                 key = "%s/%s%s" % (
279                     self.key_prefix,
280                     self.date_path,
281                     "%s/%s" % (
282                         root[len(self.root_dir):], f))
283                 key = key.strip('/')
284                 self.transfer.upload_file(
285                     os.path.join(root, f), self.bucket, key,
286                     extra_args={
287                         'ServerSideEncryption': 'AES256'})
288
```
--- 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/output.py b/c7n/output.py
--- a/c7n/output.py
+++ b/c7n/output.py
@@ -284,4 +284,5 @@
                 self.transfer.upload_file(
                     os.path.join(root, f), self.bucket, key,
                     extra_args={
+                        'ACL': 'bucket-owner-full-control',
                         'ServerSideEncryption': 'AES256'})

{"golden_diff": "diff --git a/c7n/output.py b/c7n/output.py\n--- a/c7n/output.py\n+++ b/c7n/output.py\n@@ -284,4 +284,5 @@\n                 self.transfer.upload_file(\n                     os.path.join(root, f), self.bucket, key,\n                     extra_args={\n+                        'ACL': 'bucket-owner-full-control',\n                         'ServerSideEncryption': 'AES256'})\n", "issue": "S3Output should grant bucket owner full control\nCurrently `S3Output` misses the `ACL` option. In a cross account setup it is desirable to give the bucket owner full control.\r\n\r\nWould you change the code like this?:\r\n```\r\ndiff --git a/c7n/output.py b/c7n/output.py\r\nindex c3839c2f..5fb06f59 100644\r\n--- a/c7n/output.py\r\n+++ b/c7n/output.py\r\n@@ -268,6 +268,7 @@ class S3Output(FSOutput):\r\n                 self.transfer.upload_file(\r\n                     os.path.join(root, f), self.bucket, key,\r\n                     extra_args={\r\n+                        'ACL': 'bucket-owner-full-control',\r\n                         'ServerSideEncryption': 'AES256'})\r\n \r\n     def use_s3(self):\r\n```\r\n\n", "before_files": [{"content": "# Copyright 2015-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.\n\"\"\"\nOutputs metrics, logs, structured records across\na variety of sources.\n\nSee docs/usage/outputs.rst\n\n\"\"\"\nfrom __future__ import absolute_import, division, print_function, unicode_literals\n\nimport datetime\nimport gzip\nimport logging\nimport shutil\nimport tempfile\n\nimport os\n\nfrom boto3.s3.transfer import S3Transfer\n\nfrom c7n.registry import PluginRegistry\nfrom c7n.log import CloudWatchLogHandler\nfrom c7n.utils import local_session, parse_s3, get_retry\n\nDEFAULT_NAMESPACE = \"CloudMaid\"\n\nlog = logging.getLogger('custodian.output')\n\n\nblob_outputs = PluginRegistry('c7n.blob-outputs')\n\n\nclass MetricsOutput(object):\n    \"\"\"Send metrics data to cloudwatch\n    \"\"\"\n\n    permissions = (\"cloudWatch:PutMetricData\",)\n\n    retry = staticmethod(get_retry(('Throttling',)))\n\n    @staticmethod\n    def select(metrics_enabled):\n        if metrics_enabled:\n            return MetricsOutput\n        return NullMetricsOutput\n\n    def __init__(self, ctx, namespace=DEFAULT_NAMESPACE):\n        self.ctx = ctx\n        self.namespace = namespace\n        self.buf = []\n\n    def get_timestamp(self):\n        \"\"\"\n        Now, if C7N_METRICS_TZ is set to TRUE, UTC timestamp will be used.\n        For backwards compatibility, if it is not set, UTC will be the default.\n        To disable this and use the system's time zone, C7N_METRICS_TZ shoule be set to FALSE.\n        \"\"\"\n\n        if os.getenv(\"C7N_METRICS_TZ\", '').upper() in ('TRUE', ''):\n            return datetime.datetime.utcnow()\n        else:\n            return datetime.datetime.now()\n\n    def flush(self):\n        if self.buf:\n            self._put_metrics(self.namespace, self.buf)\n            self.buf = []\n\n    def put_metric(self, key, value, unit, buffer=False, **dimensions):\n        d = {\n            \"MetricName\": key,\n            \"Timestamp\": self.get_timestamp(),\n            \"Value\": value,\n            \"Unit\": unit}\n        d[\"Dimensions\"] = [\n            {\"Name\": \"Policy\", \"Value\": self.ctx.policy.name},\n            {\"Name\": \"ResType\", \"Value\": self.ctx.policy.resource_type}]\n        for k, v in dimensions.items():\n            d['Dimensions'].append({\"Name\": k, \"Value\": v})\n\n        if buffer:\n            self.buf.append(d)\n            # Max metrics in a single request\n            if len(self.buf) == 20:\n                self.flush()\n        else:\n            self._put_metrics(self.namespace, [d])\n\n    def _put_metrics(self, ns, metrics):\n        watch = local_session(self.ctx.session_factory).client('cloudwatch')\n        return self.retry(\n            watch.put_metric_data, Namespace=ns, MetricData=metrics)\n\n\nclass NullMetricsOutput(MetricsOutput):\n\n    permissions = ()\n\n    def __init__(self, ctx, namespace=DEFAULT_NAMESPACE):\n        super(NullMetricsOutput, self).__init__(ctx, namespace)\n        self.data = []\n\n    def _put_metrics(self, ns, metrics):\n        self.data.append({'Namespace': ns, 'MetricData': metrics})\n        for m in metrics:\n            if m['MetricName'] not in ('ActionTime', 'ResourceTime'):\n                log.debug(self.format_metric(m))\n\n    def format_metric(self, m):\n        label = \"metric:%s %s:%s\" % (m['MetricName'], m['Unit'], m['Value'])\n        for d in m['Dimensions']:\n            label += \" %s:%s\" % (d['Name'].lower(), d['Value'].lower())\n        return label\n\n\nclass LogOutput(object):\n\n    log_format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'\n\n    def __init__(self, ctx):\n        self.ctx = ctx\n\n    def get_handler(self):\n        raise NotImplementedError()\n\n    def __enter__(self):\n        log.debug(\"Storing output with %s\" % repr(self))\n        self.join_log()\n        return self\n\n    def __exit__(self, exc_type=None, exc_value=None, exc_traceback=None):\n        self.leave_log()\n        if exc_type is not None:\n            log.exception(\"Error while executing policy\")\n\n    def join_log(self):\n        self.handler = self.get_handler()\n        self.handler.setLevel(logging.DEBUG)\n        self.handler.setFormatter(logging.Formatter(self.log_format))\n        mlog = logging.getLogger('custodian')\n        mlog.addHandler(self.handler)\n\n    def leave_log(self):\n        mlog = logging.getLogger('custodian')\n        mlog.removeHandler(self.handler)\n        self.handler.flush()\n        self.handler.close()\n\n\nclass CloudWatchLogOutput(LogOutput):\n\n    log_format = '%(asctime)s - %(levelname)s - %(name)s - %(message)s'\n\n    def get_handler(self):\n        return CloudWatchLogHandler(\n            log_group=self.ctx.options.log_group,\n            log_stream=self.ctx.policy.name,\n            session_factory=lambda x=None: self.ctx.session_factory(\n                assume=False))\n\n    def __repr__(self):\n        return \"<%s to group:%s stream:%s>\" % (\n            self.__class__.__name__,\n            self.ctx.options.log_group,\n            self.ctx.policy.name)\n\n\nclass FSOutput(LogOutput):\n\n    @staticmethod\n    def select(path):\n        for k in blob_outputs.keys():\n            if path.startswith('%s://' % k):\n                return blob_outputs[k]\n        # Fall back local disk\n        return blob_outputs['file']\n\n    @staticmethod\n    def join(*parts):\n        return os.path.join(*parts)\n\n    def __init__(self, ctx):\n        super(FSOutput, self).__init__(ctx)\n        self.root_dir = self.ctx.output_path or tempfile.mkdtemp()\n\n    def get_handler(self):\n        return logging.FileHandler(\n            os.path.join(self.root_dir, 'custodian-run.log'))\n\n    def compress(self):\n        # Compress files individually so thats easy to walk them, without\n        # downloading tar and extracting.\n        for root, dirs, files in os.walk(self.root_dir):\n            for f in files:\n                fp = os.path.join(root, f)\n                with gzip.open(fp + \".gz\", \"wb\", compresslevel=7) as zfh:\n                    with open(fp, \"rb\") as sfh:\n                        shutil.copyfileobj(sfh, zfh, length=2**15)\n                    os.remove(fp)\n\n\n@blob_outputs.register('file')\nclass DirectoryOutput(FSOutput):\n\n    permissions = ()\n\n    def __init__(self, ctx):\n        super(DirectoryOutput, self).__init__(ctx)\n        if self.root_dir.startswith('file://'):\n            self.root_dir = self.root_dir[len('file://'):]\n        if self.ctx.output_path is not None:\n            if not os.path.exists(self.root_dir):\n                os.makedirs(self.root_dir)\n\n    def __repr__(self):\n        return \"<%s to dir:%s>\" % (self.__class__.__name__, self.root_dir)\n\n\n@blob_outputs.register('s3')\nclass S3Output(FSOutput):\n    \"\"\"\n    Usage:\n\n    .. code-block:: python\n\n       with S3Output(session_factory, 's3://bucket/prefix'):\n           log.info('xyz')  # -> log messages sent to custodian-run.log.gz\n\n    \"\"\"\n\n    permissions = ('S3:PutObject',)\n\n    def __init__(self, ctx):\n        super(S3Output, self).__init__(ctx)\n        self.date_path = datetime.datetime.now().strftime('%Y/%m/%d/%H')\n        self.s3_path, self.bucket, self.key_prefix = parse_s3(\n            self.ctx.output_path)\n        self.root_dir = tempfile.mkdtemp()\n        self.transfer = None\n\n    def __repr__(self):\n        return \"<%s to bucket:%s prefix:%s>\" % (\n            self.__class__.__name__,\n            self.bucket,\n            \"%s/%s\" % (self.key_prefix, self.date_path))\n\n    @staticmethod\n    def join(*parts):\n        return \"/\".join([s.strip('/') for s in parts])\n\n    def __exit__(self, exc_type=None, exc_value=None, exc_traceback=None):\n        if exc_type is not None:\n            log.exception(\"Error while executing policy\")\n        log.debug(\"Uploading policy logs\")\n        self.leave_log()\n        self.compress()\n        self.transfer = S3Transfer(\n            self.ctx.session_factory(assume=False).client('s3'))\n        self.upload()\n        shutil.rmtree(self.root_dir)\n        log.debug(\"Policy Logs uploaded\")\n\n    def upload(self):\n        for root, dirs, files in os.walk(self.root_dir):\n            for f in files:\n                key = \"%s/%s%s\" % (\n                    self.key_prefix,\n                    self.date_path,\n                    \"%s/%s\" % (\n                        root[len(self.root_dir):], f))\n                key = key.strip('/')\n                self.transfer.upload_file(\n                    os.path.join(root, f), self.bucket, key,\n                    extra_args={\n                        'ServerSideEncryption': 'AES256'})\n", "path": "c7n/output.py"}], "after_files": [{"content": "# Copyright 2015-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.\n\"\"\"\nOutputs metrics, logs, structured records across\na variety of sources.\n\nSee docs/usage/outputs.rst\n\n\"\"\"\nfrom __future__ import absolute_import, division, print_function, unicode_literals\n\nimport datetime\nimport gzip\nimport logging\nimport shutil\nimport tempfile\n\nimport os\n\nfrom boto3.s3.transfer import S3Transfer\n\nfrom c7n.registry import PluginRegistry\nfrom c7n.log import CloudWatchLogHandler\nfrom c7n.utils import local_session, parse_s3, get_retry\n\nDEFAULT_NAMESPACE = \"CloudMaid\"\n\nlog = logging.getLogger('custodian.output')\n\n\nblob_outputs = PluginRegistry('c7n.blob-outputs')\n\n\nclass MetricsOutput(object):\n    \"\"\"Send metrics data to cloudwatch\n    \"\"\"\n\n    permissions = (\"cloudWatch:PutMetricData\",)\n\n    retry = staticmethod(get_retry(('Throttling',)))\n\n    @staticmethod\n    def select(metrics_enabled):\n        if metrics_enabled:\n            return MetricsOutput\n        return NullMetricsOutput\n\n    def __init__(self, ctx, namespace=DEFAULT_NAMESPACE):\n        self.ctx = ctx\n        self.namespace = namespace\n        self.buf = []\n\n    def get_timestamp(self):\n        \"\"\"\n        Now, if C7N_METRICS_TZ is set to TRUE, UTC timestamp will be used.\n        For backwards compatibility, if it is not set, UTC will be the default.\n        To disable this and use the system's time zone, C7N_METRICS_TZ shoule be set to FALSE.\n        \"\"\"\n\n        if os.getenv(\"C7N_METRICS_TZ\", '').upper() in ('TRUE', ''):\n            return datetime.datetime.utcnow()\n        else:\n            return datetime.datetime.now()\n\n    def flush(self):\n        if self.buf:\n            self._put_metrics(self.namespace, self.buf)\n            self.buf = []\n\n    def put_metric(self, key, value, unit, buffer=False, **dimensions):\n        d = {\n            \"MetricName\": key,\n            \"Timestamp\": self.get_timestamp(),\n            \"Value\": value,\n            \"Unit\": unit}\n        d[\"Dimensions\"] = [\n            {\"Name\": \"Policy\", \"Value\": self.ctx.policy.name},\n            {\"Name\": \"ResType\", \"Value\": self.ctx.policy.resource_type}]\n        for k, v in dimensions.items():\n            d['Dimensions'].append({\"Name\": k, \"Value\": v})\n\n        if buffer:\n            self.buf.append(d)\n            # Max metrics in a single request\n            if len(self.buf) == 20:\n                self.flush()\n        else:\n            self._put_metrics(self.namespace, [d])\n\n    def _put_metrics(self, ns, metrics):\n        watch = local_session(self.ctx.session_factory).client('cloudwatch')\n        return self.retry(\n            watch.put_metric_data, Namespace=ns, MetricData=metrics)\n\n\nclass NullMetricsOutput(MetricsOutput):\n\n    permissions = ()\n\n    def __init__(self, ctx, namespace=DEFAULT_NAMESPACE):\n        super(NullMetricsOutput, self).__init__(ctx, namespace)\n        self.data = []\n\n    def _put_metrics(self, ns, metrics):\n        self.data.append({'Namespace': ns, 'MetricData': metrics})\n        for m in metrics:\n            if m['MetricName'] not in ('ActionTime', 'ResourceTime'):\n                log.debug(self.format_metric(m))\n\n    def format_metric(self, m):\n        label = \"metric:%s %s:%s\" % (m['MetricName'], m['Unit'], m['Value'])\n        for d in m['Dimensions']:\n            label += \" %s:%s\" % (d['Name'].lower(), d['Value'].lower())\n        return label\n\n\nclass LogOutput(object):\n\n    log_format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'\n\n    def __init__(self, ctx):\n        self.ctx = ctx\n\n    def get_handler(self):\n        raise NotImplementedError()\n\n    def __enter__(self):\n        log.debug(\"Storing output with %s\" % repr(self))\n        self.join_log()\n        return self\n\n    def __exit__(self, exc_type=None, exc_value=None, exc_traceback=None):\n        self.leave_log()\n        if exc_type is not None:\n            log.exception(\"Error while executing policy\")\n\n    def join_log(self):\n        self.handler = self.get_handler()\n        self.handler.setLevel(logging.DEBUG)\n        self.handler.setFormatter(logging.Formatter(self.log_format))\n        mlog = logging.getLogger('custodian')\n        mlog.addHandler(self.handler)\n\n    def leave_log(self):\n        mlog = logging.getLogger('custodian')\n        mlog.removeHandler(self.handler)\n        self.handler.flush()\n        self.handler.close()\n\n\nclass CloudWatchLogOutput(LogOutput):\n\n    log_format = '%(asctime)s - %(levelname)s - %(name)s - %(message)s'\n\n    def get_handler(self):\n        return CloudWatchLogHandler(\n            log_group=self.ctx.options.log_group,\n            log_stream=self.ctx.policy.name,\n            session_factory=lambda x=None: self.ctx.session_factory(\n                assume=False))\n\n    def __repr__(self):\n        return \"<%s to group:%s stream:%s>\" % (\n            self.__class__.__name__,\n            self.ctx.options.log_group,\n            self.ctx.policy.name)\n\n\nclass FSOutput(LogOutput):\n\n    @staticmethod\n    def select(path):\n        for k in blob_outputs.keys():\n            if path.startswith('%s://' % k):\n                return blob_outputs[k]\n        # Fall back local disk\n        return blob_outputs['file']\n\n    @staticmethod\n    def join(*parts):\n        return os.path.join(*parts)\n\n    def __init__(self, ctx):\n        super(FSOutput, self).__init__(ctx)\n        self.root_dir = self.ctx.output_path or tempfile.mkdtemp()\n\n    def get_handler(self):\n        return logging.FileHandler(\n            os.path.join(self.root_dir, 'custodian-run.log'))\n\n    def compress(self):\n        # Compress files individually so thats easy to walk them, without\n        # downloading tar and extracting.\n        for root, dirs, files in os.walk(self.root_dir):\n            for f in files:\n                fp = os.path.join(root, f)\n                with gzip.open(fp + \".gz\", \"wb\", compresslevel=7) as zfh:\n                    with open(fp, \"rb\") as sfh:\n                        shutil.copyfileobj(sfh, zfh, length=2**15)\n                    os.remove(fp)\n\n\n@blob_outputs.register('file')\nclass DirectoryOutput(FSOutput):\n\n    permissions = ()\n\n    def __init__(self, ctx):\n        super(DirectoryOutput, self).__init__(ctx)\n        if self.root_dir.startswith('file://'):\n            self.root_dir = self.root_dir[len('file://'):]\n        if self.ctx.output_path is not None:\n            if not os.path.exists(self.root_dir):\n                os.makedirs(self.root_dir)\n\n    def __repr__(self):\n        return \"<%s to dir:%s>\" % (self.__class__.__name__, self.root_dir)\n\n\n@blob_outputs.register('s3')\nclass S3Output(FSOutput):\n    \"\"\"\n    Usage:\n\n    .. code-block:: python\n\n       with S3Output(session_factory, 's3://bucket/prefix'):\n           log.info('xyz')  # -> log messages sent to custodian-run.log.gz\n\n    \"\"\"\n\n    permissions = ('S3:PutObject',)\n\n    def __init__(self, ctx):\n        super(S3Output, self).__init__(ctx)\n        self.date_path = datetime.datetime.now().strftime('%Y/%m/%d/%H')\n        self.s3_path, self.bucket, self.key_prefix = parse_s3(\n            self.ctx.output_path)\n        self.root_dir = tempfile.mkdtemp()\n        self.transfer = None\n\n    def __repr__(self):\n        return \"<%s to bucket:%s prefix:%s>\" % (\n            self.__class__.__name__,\n            self.bucket,\n            \"%s/%s\" % (self.key_prefix, self.date_path))\n\n    @staticmethod\n    def join(*parts):\n        return \"/\".join([s.strip('/') for s in parts])\n\n    def __exit__(self, exc_type=None, exc_value=None, exc_traceback=None):\n        if exc_type is not None:\n            log.exception(\"Error while executing policy\")\n        log.debug(\"Uploading policy logs\")\n        self.leave_log()\n        self.compress()\n        self.transfer = S3Transfer(\n            self.ctx.session_factory(assume=False).client('s3'))\n        self.upload()\n        shutil.rmtree(self.root_dir)\n        log.debug(\"Policy Logs uploaded\")\n\n    def upload(self):\n        for root, dirs, files in os.walk(self.root_dir):\n            for f in files:\n                key = \"%s/%s%s\" % (\n                    self.key_prefix,\n                    self.date_path,\n                    \"%s/%s\" % (\n                        root[len(self.root_dir):], f))\n                key = key.strip('/')\n                self.transfer.upload_file(\n                    os.path.join(root, f), self.bucket, key,\n                    extra_args={\n                        'ACL': 'bucket-owner-full-control',\n                        'ServerSideEncryption': 'AES256'})\n", "path": "c7n/output.py"}]}