We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
google.cloud.logging.handlers: send messages to stderr
Would it be ok if we do the print at https://github.com/GoogleCloudPlatform/google-cloud-python/blob/master/logging/google/cloud/logging/handlers/transports/background_thread.py#L222
to stderr instead? So it doesn't disturb the output of the software?

Thanks

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `logging/google/cloud/logging/handlers/transports/background_thread.py`
Content:
```
1 # Copyright 2016 Google 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 """Transport for Python logging handler
16 
17 Uses a background worker to log to Stackdriver Logging asynchronously.
18 """
19 
20 from __future__ import print_function
21 
22 import atexit
23 import logging
24 import threading
25 import time
26 
27 from six.moves import range
28 from six.moves import queue
29 
30 from google.cloud.logging.handlers.transports.base import Transport
31 
32 _DEFAULT_GRACE_PERIOD = 5.0  # Seconds
33 _DEFAULT_MAX_BATCH_SIZE = 10
34 _DEFAULT_MAX_LATENCY = 0  # Seconds
35 _WORKER_THREAD_NAME = 'google.cloud.logging.Worker'
36 _WORKER_TERMINATOR = object()
37 _LOGGER = logging.getLogger(__name__)
38 
39 
40 def _get_many(queue_, max_items=None, max_latency=0):
41     """Get multiple items from a Queue.
42 
43     Gets at least one (blocking) and at most ``max_items`` items
44     (non-blocking) from a given Queue. Does not mark the items as done.
45 
46     :type queue_: :class:`~queue.Queue`
47     :param queue_: The Queue to get items from.
48 
49     :type max_items: int
50     :param max_items: The maximum number of items to get. If ``None``, then all
51         available items in the queue are returned.
52 
53     :type max_latency: float
54     :param max_latency: The maximum number of seconds to wait for more than one
55         item from a queue. This number includes the time required to retrieve
56         the first item.
57 
58     :rtype: Sequence
59     :returns: A sequence of items retrieved from the queue.
60     """
61     start = time.time()
62     # Always return at least one item.
63     items = [queue_.get()]
64     while max_items is None or len(items) < max_items:
65         try:
66             elapsed = time.time() - start
67             timeout = max(0, max_latency - elapsed)
68             items.append(queue_.get(timeout=timeout))
69         except queue.Empty:
70             break
71     return items
72 
73 
74 class _Worker(object):
75     """A background thread that writes batches of log entries.
76 
77     :type cloud_logger: :class:`~google.cloud.logging.logger.Logger`
78     :param cloud_logger: The logger to send entries to.
79 
80     :type grace_period: float
81     :param grace_period: The amount of time to wait for pending logs to
82         be submitted when the process is shutting down.
83 
84     :type max_batch_size: int
85     :param max_batch_size: The maximum number of items to send at a time
86         in the background thread.
87 
88     :type max_latency: float
89     :param max_latency: The amount of time to wait for new logs before
90         sending a new batch. It is strongly recommended to keep this smaller
91         than the grace_period. This means this is effectively the longest
92         amount of time the background thread will hold onto log entries
93         before sending them to the server.
94     """
95 
96     def __init__(self, cloud_logger, grace_period=_DEFAULT_GRACE_PERIOD,
97                  max_batch_size=_DEFAULT_MAX_BATCH_SIZE,
98                  max_latency=_DEFAULT_MAX_LATENCY):
99         self._cloud_logger = cloud_logger
100         self._grace_period = grace_period
101         self._max_batch_size = max_batch_size
102         self._max_latency = max_latency
103         self._queue = queue.Queue(0)
104         self._operational_lock = threading.Lock()
105         self._thread = None
106 
107     @property
108     def is_alive(self):
109         """Returns True is the background thread is running."""
110         return self._thread is not None and self._thread.is_alive()
111 
112     def _safely_commit_batch(self, batch):
113         total_logs = len(batch.entries)
114 
115         try:
116             if total_logs > 0:
117                 batch.commit()
118                 _LOGGER.debug('Submitted %d logs', total_logs)
119         except Exception:
120             _LOGGER.error(
121                 'Failed to submit %d logs.', total_logs, exc_info=True)
122 
123     def _thread_main(self):
124         """The entry point for the worker thread.
125 
126         Pulls pending log entries off the queue and writes them in batches to
127         the Cloud Logger.
128         """
129         _LOGGER.debug('Background thread started.')
130 
131         quit_ = False
132         while True:
133             batch = self._cloud_logger.batch()
134             items = _get_many(
135                 self._queue, max_items=self._max_batch_size,
136                 max_latency=self._max_latency)
137 
138             for item in items:
139                 if item is _WORKER_TERMINATOR:
140                     quit_ = True
141                     # Continue processing items, don't break, try to process
142                     # all items we got back before quitting.
143                 else:
144                     batch.log_struct(**item)
145 
146             self._safely_commit_batch(batch)
147 
148             for _ in range(len(items)):
149                 self._queue.task_done()
150 
151             if quit_:
152                 break
153 
154         _LOGGER.debug('Background thread exited gracefully.')
155 
156     def start(self):
157         """Starts the background thread.
158 
159         Additionally, this registers a handler for process exit to attempt
160         to send any pending log entries before shutdown.
161         """
162         with self._operational_lock:
163             if self.is_alive:
164                 return
165 
166             self._thread = threading.Thread(
167                 target=self._thread_main,
168                 name=_WORKER_THREAD_NAME)
169             self._thread.daemon = True
170             self._thread.start()
171             atexit.register(self._main_thread_terminated)
172 
173     def stop(self, grace_period=None):
174         """Signals the background thread to stop.
175 
176         This does not terminate the background thread. It simply queues the
177         stop signal. If the main process exits before the background thread
178         processes the stop signal, it will be terminated without finishing
179         work. The ``grace_period`` parameter will give the background
180         thread some time to finish processing before this function returns.
181 
182         :type grace_period: float
183         :param grace_period: If specified, this method will block up to this
184             many seconds to allow the background thread to finish work before
185             returning.
186 
187         :rtype: bool
188         :returns: True if the thread terminated. False if the thread is still
189             running.
190         """
191         if not self.is_alive:
192             return True
193 
194         with self._operational_lock:
195             self._queue.put_nowait(_WORKER_TERMINATOR)
196 
197             if grace_period is not None:
198                 print('Waiting up to %d seconds.' % (grace_period,))
199 
200             self._thread.join(timeout=grace_period)
201 
202             # Check this before disowning the thread, because after we disown
203             # the thread is_alive will be False regardless of if the thread
204             # exited or not.
205             success = not self.is_alive
206 
207             self._thread = None
208 
209             return success
210 
211     def _main_thread_terminated(self):
212         """Callback that attempts to send pending logs before termination."""
213         if not self.is_alive:
214             return
215 
216         if not self._queue.empty():
217             print(
218                 'Program shutting down, attempting to send %d queued log '
219                 'entries to Stackdriver Logging...' % (self._queue.qsize(),))
220 
221         if self.stop(self._grace_period):
222             print('Sent all pending logs.')
223         else:
224             print('Failed to send %d pending logs.' % (self._queue.qsize(),))
225 
226     def enqueue(self, record, message, resource=None, labels=None):
227         """Queues a log entry to be written by the background thread.
228 
229         :type record: :class:`logging.LogRecord`
230         :param record: Python log record that the handler was called with.
231 
232         :type message: str
233         :param message: The message from the ``LogRecord`` after being
234                         formatted by the associated log formatters.
235 
236         :type resource: :class:`~google.cloud.logging.resource.Resource`
237         :param resource: (Optional) Monitored resource of the entry
238 
239         :type labels: dict
240         :param labels: (Optional) Mapping of labels for the entry.
241         """
242         self._queue.put_nowait({
243             'info': {
244                 'message': message,
245                 'python_logger': record.name,
246             },
247             'severity': record.levelname,
248             'resource': resource,
249             'labels': labels,
250         })
251 
252     def flush(self):
253         """Submit any pending log records."""
254         self._queue.join()
255 
256 
257 class BackgroundThreadTransport(Transport):
258     """Asynchronous transport that uses a background thread.
259 
260     :type client: :class:`~google.cloud.logging.client.Client`
261     :param client: The Logging client.
262 
263     :type name: str
264     :param name: the name of the logger.
265 
266     :type grace_period: float
267     :param grace_period: The amount of time to wait for pending logs to
268         be submitted when the process is shutting down.
269 
270     :type batch_size: int
271     :param batch_size: The maximum number of items to send at a time in the
272         background thread.
273 
274     :type max_latency: float
275     :param max_latency: The amount of time to wait for new logs before
276         sending a new batch. It is strongly recommended to keep this smaller
277         than the grace_period. This means this is effectively the longest
278         amount of time the background thread will hold onto log entries
279         before sending them to the server.
280     """
281 
282     def __init__(self, client, name, grace_period=_DEFAULT_GRACE_PERIOD,
283                  batch_size=_DEFAULT_MAX_BATCH_SIZE,
284                  max_latency=_DEFAULT_MAX_LATENCY):
285         self.client = client
286         logger = self.client.logger(name)
287         self.worker = _Worker(logger,
288                               grace_period=grace_period,
289                               max_batch_size=batch_size,
290                               max_latency=max_latency)
291         self.worker.start()
292 
293     def send(self, record, message, resource=None, labels=None):
294         """Overrides Transport.send().
295 
296         :type record: :class:`logging.LogRecord`
297         :param record: Python log record that the handler was called with.
298 
299         :type message: str
300         :param message: The message from the ``LogRecord`` after being
301                         formatted by the associated log formatters.
302 
303         :type resource: :class:`~google.cloud.logging.resource.Resource`
304         :param resource: (Optional) Monitored resource of the entry.
305 
306         :type labels: dict
307         :param labels: (Optional) Mapping of labels for the entry.
308         """
309         self.worker.enqueue(record, message, resource=resource, labels=labels)
310 
311     def flush(self):
312         """Submit any pending log records."""
313         self.worker.flush()
314
```
--- 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/logging/google/cloud/logging/handlers/transports/background_thread.py b/logging/google/cloud/logging/handlers/transports/background_thread.py
--- a/logging/google/cloud/logging/handlers/transports/background_thread.py
+++ b/logging/google/cloud/logging/handlers/transports/background_thread.py
@@ -21,6 +21,7 @@
 
 import atexit
 import logging
+import sys
 import threading
 import time
 
@@ -195,7 +196,9 @@
             self._queue.put_nowait(_WORKER_TERMINATOR)
 
             if grace_period is not None:
-                print('Waiting up to %d seconds.' % (grace_period,))
+                print(
+                    'Waiting up to %d seconds.' % (grace_period,),
+                    file=sys.stderr)
 
             self._thread.join(timeout=grace_period)
 
@@ -216,12 +219,15 @@
         if not self._queue.empty():
             print(
                 'Program shutting down, attempting to send %d queued log '
-                'entries to Stackdriver Logging...' % (self._queue.qsize(),))
+                'entries to Stackdriver Logging...' % (self._queue.qsize(),),
+                file=sys.stderr)
 
         if self.stop(self._grace_period):
-            print('Sent all pending logs.')
+            print('Sent all pending logs.', file=sys.stderr)
         else:
-            print('Failed to send %d pending logs.' % (self._queue.qsize(),))
+            print(
+                'Failed to send %d pending logs.' % (self._queue.qsize(),),
+                file=sys.stderr)
 
     def enqueue(self, record, message, resource=None, labels=None):
         """Queues a log entry to be written by the background thread.

{"golden_diff": "diff --git a/logging/google/cloud/logging/handlers/transports/background_thread.py b/logging/google/cloud/logging/handlers/transports/background_thread.py\n--- a/logging/google/cloud/logging/handlers/transports/background_thread.py\n+++ b/logging/google/cloud/logging/handlers/transports/background_thread.py\n@@ -21,6 +21,7 @@\n \n import atexit\n import logging\n+import sys\n import threading\n import time\n \n@@ -195,7 +196,9 @@\n             self._queue.put_nowait(_WORKER_TERMINATOR)\n \n             if grace_period is not None:\n-                print('Waiting up to %d seconds.' % (grace_period,))\n+                print(\n+                    'Waiting up to %d seconds.' % (grace_period,),\n+                    file=sys.stderr)\n \n             self._thread.join(timeout=grace_period)\n \n@@ -216,12 +219,15 @@\n         if not self._queue.empty():\n             print(\n                 'Program shutting down, attempting to send %d queued log '\n-                'entries to Stackdriver Logging...' % (self._queue.qsize(),))\n+                'entries to Stackdriver Logging...' % (self._queue.qsize(),),\n+                file=sys.stderr)\n \n         if self.stop(self._grace_period):\n-            print('Sent all pending logs.')\n+            print('Sent all pending logs.', file=sys.stderr)\n         else:\n-            print('Failed to send %d pending logs.' % (self._queue.qsize(),))\n+            print(\n+                'Failed to send %d pending logs.' % (self._queue.qsize(),),\n+                file=sys.stderr)\n \n     def enqueue(self, record, message, resource=None, labels=None):\n         \"\"\"Queues a log entry to be written by the background thread.\n", "issue": "google.cloud.logging.handlers: send messages to stderr\nWould it be ok if we do the print at https://github.com/GoogleCloudPlatform/google-cloud-python/blob/master/logging/google/cloud/logging/handlers/transports/background_thread.py#L222\r\nto stderr instead? So it doesn't disturb the output of the software?\r\n\r\nThanks\n", "before_files": [{"content": "# Copyright 2016 Google 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\n\"\"\"Transport for Python logging handler\n\nUses a background worker to log to Stackdriver Logging asynchronously.\n\"\"\"\n\nfrom __future__ import print_function\n\nimport atexit\nimport logging\nimport threading\nimport time\n\nfrom six.moves import range\nfrom six.moves import queue\n\nfrom google.cloud.logging.handlers.transports.base import Transport\n\n_DEFAULT_GRACE_PERIOD = 5.0  # Seconds\n_DEFAULT_MAX_BATCH_SIZE = 10\n_DEFAULT_MAX_LATENCY = 0  # Seconds\n_WORKER_THREAD_NAME = 'google.cloud.logging.Worker'\n_WORKER_TERMINATOR = object()\n_LOGGER = logging.getLogger(__name__)\n\n\ndef _get_many(queue_, max_items=None, max_latency=0):\n    \"\"\"Get multiple items from a Queue.\n\n    Gets at least one (blocking) and at most ``max_items`` items\n    (non-blocking) from a given Queue. Does not mark the items as done.\n\n    :type queue_: :class:`~queue.Queue`\n    :param queue_: The Queue to get items from.\n\n    :type max_items: int\n    :param max_items: The maximum number of items to get. If ``None``, then all\n        available items in the queue are returned.\n\n    :type max_latency: float\n    :param max_latency: The maximum number of seconds to wait for more than one\n        item from a queue. This number includes the time required to retrieve\n        the first item.\n\n    :rtype: Sequence\n    :returns: A sequence of items retrieved from the queue.\n    \"\"\"\n    start = time.time()\n    # Always return at least one item.\n    items = [queue_.get()]\n    while max_items is None or len(items) < max_items:\n        try:\n            elapsed = time.time() - start\n            timeout = max(0, max_latency - elapsed)\n            items.append(queue_.get(timeout=timeout))\n        except queue.Empty:\n            break\n    return items\n\n\nclass _Worker(object):\n    \"\"\"A background thread that writes batches of log entries.\n\n    :type cloud_logger: :class:`~google.cloud.logging.logger.Logger`\n    :param cloud_logger: The logger to send entries to.\n\n    :type grace_period: float\n    :param grace_period: The amount of time to wait for pending logs to\n        be submitted when the process is shutting down.\n\n    :type max_batch_size: int\n    :param max_batch_size: The maximum number of items to send at a time\n        in the background thread.\n\n    :type max_latency: float\n    :param max_latency: The amount of time to wait for new logs before\n        sending a new batch. It is strongly recommended to keep this smaller\n        than the grace_period. This means this is effectively the longest\n        amount of time the background thread will hold onto log entries\n        before sending them to the server.\n    \"\"\"\n\n    def __init__(self, cloud_logger, grace_period=_DEFAULT_GRACE_PERIOD,\n                 max_batch_size=_DEFAULT_MAX_BATCH_SIZE,\n                 max_latency=_DEFAULT_MAX_LATENCY):\n        self._cloud_logger = cloud_logger\n        self._grace_period = grace_period\n        self._max_batch_size = max_batch_size\n        self._max_latency = max_latency\n        self._queue = queue.Queue(0)\n        self._operational_lock = threading.Lock()\n        self._thread = None\n\n    @property\n    def is_alive(self):\n        \"\"\"Returns True is the background thread is running.\"\"\"\n        return self._thread is not None and self._thread.is_alive()\n\n    def _safely_commit_batch(self, batch):\n        total_logs = len(batch.entries)\n\n        try:\n            if total_logs > 0:\n                batch.commit()\n                _LOGGER.debug('Submitted %d logs', total_logs)\n        except Exception:\n            _LOGGER.error(\n                'Failed to submit %d logs.', total_logs, exc_info=True)\n\n    def _thread_main(self):\n        \"\"\"The entry point for the worker thread.\n\n        Pulls pending log entries off the queue and writes them in batches to\n        the Cloud Logger.\n        \"\"\"\n        _LOGGER.debug('Background thread started.')\n\n        quit_ = False\n        while True:\n            batch = self._cloud_logger.batch()\n            items = _get_many(\n                self._queue, max_items=self._max_batch_size,\n                max_latency=self._max_latency)\n\n            for item in items:\n                if item is _WORKER_TERMINATOR:\n                    quit_ = True\n                    # Continue processing items, don't break, try to process\n                    # all items we got back before quitting.\n                else:\n                    batch.log_struct(**item)\n\n            self._safely_commit_batch(batch)\n\n            for _ in range(len(items)):\n                self._queue.task_done()\n\n            if quit_:\n                break\n\n        _LOGGER.debug('Background thread exited gracefully.')\n\n    def start(self):\n        \"\"\"Starts the background thread.\n\n        Additionally, this registers a handler for process exit to attempt\n        to send any pending log entries before shutdown.\n        \"\"\"\n        with self._operational_lock:\n            if self.is_alive:\n                return\n\n            self._thread = threading.Thread(\n                target=self._thread_main,\n                name=_WORKER_THREAD_NAME)\n            self._thread.daemon = True\n            self._thread.start()\n            atexit.register(self._main_thread_terminated)\n\n    def stop(self, grace_period=None):\n        \"\"\"Signals the background thread to stop.\n\n        This does not terminate the background thread. It simply queues the\n        stop signal. If the main process exits before the background thread\n        processes the stop signal, it will be terminated without finishing\n        work. The ``grace_period`` parameter will give the background\n        thread some time to finish processing before this function returns.\n\n        :type grace_period: float\n        :param grace_period: If specified, this method will block up to this\n            many seconds to allow the background thread to finish work before\n            returning.\n\n        :rtype: bool\n        :returns: True if the thread terminated. False if the thread is still\n            running.\n        \"\"\"\n        if not self.is_alive:\n            return True\n\n        with self._operational_lock:\n            self._queue.put_nowait(_WORKER_TERMINATOR)\n\n            if grace_period is not None:\n                print('Waiting up to %d seconds.' % (grace_period,))\n\n            self._thread.join(timeout=grace_period)\n\n            # Check this before disowning the thread, because after we disown\n            # the thread is_alive will be False regardless of if the thread\n            # exited or not.\n            success = not self.is_alive\n\n            self._thread = None\n\n            return success\n\n    def _main_thread_terminated(self):\n        \"\"\"Callback that attempts to send pending logs before termination.\"\"\"\n        if not self.is_alive:\n            return\n\n        if not self._queue.empty():\n            print(\n                'Program shutting down, attempting to send %d queued log '\n                'entries to Stackdriver Logging...' % (self._queue.qsize(),))\n\n        if self.stop(self._grace_period):\n            print('Sent all pending logs.')\n        else:\n            print('Failed to send %d pending logs.' % (self._queue.qsize(),))\n\n    def enqueue(self, record, message, resource=None, labels=None):\n        \"\"\"Queues a log entry to be written by the background thread.\n\n        :type record: :class:`logging.LogRecord`\n        :param record: Python log record that the handler was called with.\n\n        :type message: str\n        :param message: The message from the ``LogRecord`` after being\n                        formatted by the associated log formatters.\n\n        :type resource: :class:`~google.cloud.logging.resource.Resource`\n        :param resource: (Optional) Monitored resource of the entry\n\n        :type labels: dict\n        :param labels: (Optional) Mapping of labels for the entry.\n        \"\"\"\n        self._queue.put_nowait({\n            'info': {\n                'message': message,\n                'python_logger': record.name,\n            },\n            'severity': record.levelname,\n            'resource': resource,\n            'labels': labels,\n        })\n\n    def flush(self):\n        \"\"\"Submit any pending log records.\"\"\"\n        self._queue.join()\n\n\nclass BackgroundThreadTransport(Transport):\n    \"\"\"Asynchronous transport that uses a background thread.\n\n    :type client: :class:`~google.cloud.logging.client.Client`\n    :param client: The Logging client.\n\n    :type name: str\n    :param name: the name of the logger.\n\n    :type grace_period: float\n    :param grace_period: The amount of time to wait for pending logs to\n        be submitted when the process is shutting down.\n\n    :type batch_size: int\n    :param batch_size: The maximum number of items to send at a time in the\n        background thread.\n\n    :type max_latency: float\n    :param max_latency: The amount of time to wait for new logs before\n        sending a new batch. It is strongly recommended to keep this smaller\n        than the grace_period. This means this is effectively the longest\n        amount of time the background thread will hold onto log entries\n        before sending them to the server.\n    \"\"\"\n\n    def __init__(self, client, name, grace_period=_DEFAULT_GRACE_PERIOD,\n                 batch_size=_DEFAULT_MAX_BATCH_SIZE,\n                 max_latency=_DEFAULT_MAX_LATENCY):\n        self.client = client\n        logger = self.client.logger(name)\n        self.worker = _Worker(logger,\n                              grace_period=grace_period,\n                              max_batch_size=batch_size,\n                              max_latency=max_latency)\n        self.worker.start()\n\n    def send(self, record, message, resource=None, labels=None):\n        \"\"\"Overrides Transport.send().\n\n        :type record: :class:`logging.LogRecord`\n        :param record: Python log record that the handler was called with.\n\n        :type message: str\n        :param message: The message from the ``LogRecord`` after being\n                        formatted by the associated log formatters.\n\n        :type resource: :class:`~google.cloud.logging.resource.Resource`\n        :param resource: (Optional) Monitored resource of the entry.\n\n        :type labels: dict\n        :param labels: (Optional) Mapping of labels for the entry.\n        \"\"\"\n        self.worker.enqueue(record, message, resource=resource, labels=labels)\n\n    def flush(self):\n        \"\"\"Submit any pending log records.\"\"\"\n        self.worker.flush()\n", "path": "logging/google/cloud/logging/handlers/transports/background_thread.py"}], "after_files": [{"content": "# Copyright 2016 Google 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\n\"\"\"Transport for Python logging handler\n\nUses a background worker to log to Stackdriver Logging asynchronously.\n\"\"\"\n\nfrom __future__ import print_function\n\nimport atexit\nimport logging\nimport sys\nimport threading\nimport time\n\nfrom six.moves import range\nfrom six.moves import queue\n\nfrom google.cloud.logging.handlers.transports.base import Transport\n\n_DEFAULT_GRACE_PERIOD = 5.0  # Seconds\n_DEFAULT_MAX_BATCH_SIZE = 10\n_DEFAULT_MAX_LATENCY = 0  # Seconds\n_WORKER_THREAD_NAME = 'google.cloud.logging.Worker'\n_WORKER_TERMINATOR = object()\n_LOGGER = logging.getLogger(__name__)\n\n\ndef _get_many(queue_, max_items=None, max_latency=0):\n    \"\"\"Get multiple items from a Queue.\n\n    Gets at least one (blocking) and at most ``max_items`` items\n    (non-blocking) from a given Queue. Does not mark the items as done.\n\n    :type queue_: :class:`~queue.Queue`\n    :param queue_: The Queue to get items from.\n\n    :type max_items: int\n    :param max_items: The maximum number of items to get. If ``None``, then all\n        available items in the queue are returned.\n\n    :type max_latency: float\n    :param max_latency: The maximum number of seconds to wait for more than one\n        item from a queue. This number includes the time required to retrieve\n        the first item.\n\n    :rtype: Sequence\n    :returns: A sequence of items retrieved from the queue.\n    \"\"\"\n    start = time.time()\n    # Always return at least one item.\n    items = [queue_.get()]\n    while max_items is None or len(items) < max_items:\n        try:\n            elapsed = time.time() - start\n            timeout = max(0, max_latency - elapsed)\n            items.append(queue_.get(timeout=timeout))\n        except queue.Empty:\n            break\n    return items\n\n\nclass _Worker(object):\n    \"\"\"A background thread that writes batches of log entries.\n\n    :type cloud_logger: :class:`~google.cloud.logging.logger.Logger`\n    :param cloud_logger: The logger to send entries to.\n\n    :type grace_period: float\n    :param grace_period: The amount of time to wait for pending logs to\n        be submitted when the process is shutting down.\n\n    :type max_batch_size: int\n    :param max_batch_size: The maximum number of items to send at a time\n        in the background thread.\n\n    :type max_latency: float\n    :param max_latency: The amount of time to wait for new logs before\n        sending a new batch. It is strongly recommended to keep this smaller\n        than the grace_period. This means this is effectively the longest\n        amount of time the background thread will hold onto log entries\n        before sending them to the server.\n    \"\"\"\n\n    def __init__(self, cloud_logger, grace_period=_DEFAULT_GRACE_PERIOD,\n                 max_batch_size=_DEFAULT_MAX_BATCH_SIZE,\n                 max_latency=_DEFAULT_MAX_LATENCY):\n        self._cloud_logger = cloud_logger\n        self._grace_period = grace_period\n        self._max_batch_size = max_batch_size\n        self._max_latency = max_latency\n        self._queue = queue.Queue(0)\n        self._operational_lock = threading.Lock()\n        self._thread = None\n\n    @property\n    def is_alive(self):\n        \"\"\"Returns True is the background thread is running.\"\"\"\n        return self._thread is not None and self._thread.is_alive()\n\n    def _safely_commit_batch(self, batch):\n        total_logs = len(batch.entries)\n\n        try:\n            if total_logs > 0:\n                batch.commit()\n                _LOGGER.debug('Submitted %d logs', total_logs)\n        except Exception:\n            _LOGGER.error(\n                'Failed to submit %d logs.', total_logs, exc_info=True)\n\n    def _thread_main(self):\n        \"\"\"The entry point for the worker thread.\n\n        Pulls pending log entries off the queue and writes them in batches to\n        the Cloud Logger.\n        \"\"\"\n        _LOGGER.debug('Background thread started.')\n\n        quit_ = False\n        while True:\n            batch = self._cloud_logger.batch()\n            items = _get_many(\n                self._queue, max_items=self._max_batch_size,\n                max_latency=self._max_latency)\n\n            for item in items:\n                if item is _WORKER_TERMINATOR:\n                    quit_ = True\n                    # Continue processing items, don't break, try to process\n                    # all items we got back before quitting.\n                else:\n                    batch.log_struct(**item)\n\n            self._safely_commit_batch(batch)\n\n            for _ in range(len(items)):\n                self._queue.task_done()\n\n            if quit_:\n                break\n\n        _LOGGER.debug('Background thread exited gracefully.')\n\n    def start(self):\n        \"\"\"Starts the background thread.\n\n        Additionally, this registers a handler for process exit to attempt\n        to send any pending log entries before shutdown.\n        \"\"\"\n        with self._operational_lock:\n            if self.is_alive:\n                return\n\n            self._thread = threading.Thread(\n                target=self._thread_main,\n                name=_WORKER_THREAD_NAME)\n            self._thread.daemon = True\n            self._thread.start()\n            atexit.register(self._main_thread_terminated)\n\n    def stop(self, grace_period=None):\n        \"\"\"Signals the background thread to stop.\n\n        This does not terminate the background thread. It simply queues the\n        stop signal. If the main process exits before the background thread\n        processes the stop signal, it will be terminated without finishing\n        work. The ``grace_period`` parameter will give the background\n        thread some time to finish processing before this function returns.\n\n        :type grace_period: float\n        :param grace_period: If specified, this method will block up to this\n            many seconds to allow the background thread to finish work before\n            returning.\n\n        :rtype: bool\n        :returns: True if the thread terminated. False if the thread is still\n            running.\n        \"\"\"\n        if not self.is_alive:\n            return True\n\n        with self._operational_lock:\n            self._queue.put_nowait(_WORKER_TERMINATOR)\n\n            if grace_period is not None:\n                print(\n                    'Waiting up to %d seconds.' % (grace_period,),\n                    file=sys.stderr)\n\n            self._thread.join(timeout=grace_period)\n\n            # Check this before disowning the thread, because after we disown\n            # the thread is_alive will be False regardless of if the thread\n            # exited or not.\n            success = not self.is_alive\n\n            self._thread = None\n\n            return success\n\n    def _main_thread_terminated(self):\n        \"\"\"Callback that attempts to send pending logs before termination.\"\"\"\n        if not self.is_alive:\n            return\n\n        if not self._queue.empty():\n            print(\n                'Program shutting down, attempting to send %d queued log '\n                'entries to Stackdriver Logging...' % (self._queue.qsize(),),\n                file=sys.stderr)\n\n        if self.stop(self._grace_period):\n            print('Sent all pending logs.', file=sys.stderr)\n        else:\n            print(\n                'Failed to send %d pending logs.' % (self._queue.qsize(),),\n                file=sys.stderr)\n\n    def enqueue(self, record, message, resource=None, labels=None):\n        \"\"\"Queues a log entry to be written by the background thread.\n\n        :type record: :class:`logging.LogRecord`\n        :param record: Python log record that the handler was called with.\n\n        :type message: str\n        :param message: The message from the ``LogRecord`` after being\n                        formatted by the associated log formatters.\n\n        :type resource: :class:`~google.cloud.logging.resource.Resource`\n        :param resource: (Optional) Monitored resource of the entry\n\n        :type labels: dict\n        :param labels: (Optional) Mapping of labels for the entry.\n        \"\"\"\n        self._queue.put_nowait({\n            'info': {\n                'message': message,\n                'python_logger': record.name,\n            },\n            'severity': record.levelname,\n            'resource': resource,\n            'labels': labels,\n        })\n\n    def flush(self):\n        \"\"\"Submit any pending log records.\"\"\"\n        self._queue.join()\n\n\nclass BackgroundThreadTransport(Transport):\n    \"\"\"Asynchronous transport that uses a background thread.\n\n    :type client: :class:`~google.cloud.logging.client.Client`\n    :param client: The Logging client.\n\n    :type name: str\n    :param name: the name of the logger.\n\n    :type grace_period: float\n    :param grace_period: The amount of time to wait for pending logs to\n        be submitted when the process is shutting down.\n\n    :type batch_size: int\n    :param batch_size: The maximum number of items to send at a time in the\n        background thread.\n\n    :type max_latency: float\n    :param max_latency: The amount of time to wait for new logs before\n        sending a new batch. It is strongly recommended to keep this smaller\n        than the grace_period. This means this is effectively the longest\n        amount of time the background thread will hold onto log entries\n        before sending them to the server.\n    \"\"\"\n\n    def __init__(self, client, name, grace_period=_DEFAULT_GRACE_PERIOD,\n                 batch_size=_DEFAULT_MAX_BATCH_SIZE,\n                 max_latency=_DEFAULT_MAX_LATENCY):\n        self.client = client\n        logger = self.client.logger(name)\n        self.worker = _Worker(logger,\n                              grace_period=grace_period,\n                              max_batch_size=batch_size,\n                              max_latency=max_latency)\n        self.worker.start()\n\n    def send(self, record, message, resource=None, labels=None):\n        \"\"\"Overrides Transport.send().\n\n        :type record: :class:`logging.LogRecord`\n        :param record: Python log record that the handler was called with.\n\n        :type message: str\n        :param message: The message from the ``LogRecord`` after being\n                        formatted by the associated log formatters.\n\n        :type resource: :class:`~google.cloud.logging.resource.Resource`\n        :param resource: (Optional) Monitored resource of the entry.\n\n        :type labels: dict\n        :param labels: (Optional) Mapping of labels for the entry.\n        \"\"\"\n        self.worker.enqueue(record, message, resource=resource, labels=labels)\n\n    def flush(self):\n        \"\"\"Submit any pending log records.\"\"\"\n        self.worker.flush()\n", "path": "logging/google/cloud/logging/handlers/transports/background_thread.py"}]}