Implement flush for cloud logging handlers (googleapis#3413)

Author: Jon Wayne Parrott

logging/google/cloud/logging/handlers/transports/background_thread.py

@@ -221,6 +221,10 @@ def enqueue(self, record, message):
             'severity': record.levelname,
         })
 
+    def flush(self):
+        """Submit any pending log records."""
+        self._queue.join()
+
 
 class BackgroundThreadTransport(Transport):
     """Asynchronous transport that uses a background thread.
@@ -260,3 +264,7 @@ def send(self, record, message):
                         formatted by the associated log formatters.
         """
         self.worker.enqueue(record, message)
+
+    def flush(self):
+        """Submit any pending log records."""
+        self.worker.flush()

logging/google/cloud/logging/handlers/transports/base.py

@@ -33,3 +33,9 @@ def send(self, record, message):
                         formatted by the associated log formatters.
         """
         raise NotImplementedError
+
+    def flush(self):
+        """Submit any pending log records.
+
+        For blocking/sync transports, this is a no-op.
+        """

logging/tests/system.py

@@ -247,6 +247,7 @@ def test_log_handler_async(self):
         cloud_logger = logging.getLogger(handler.name)
         cloud_logger.addHandler(handler)
         cloud_logger.warn(LOG_MESSAGE)
+        handler.flush()
         entries = _list_entries(logger)
         expected_payload = {
             'message': LOG_MESSAGE,

logging/tests/unit/handlers/transports/test_background_thread.py

@@ -16,6 +16,7 @@
 import unittest
 
 import mock
+from six.moves import queue
 
 
 class TestBackgroundThreadHandler(unittest.TestCase):
@@ -61,6 +62,16 @@ def test_send(self):
 
         transport.worker.enqueue.assert_called_once_with(record, message)
 
+    def test_flush(self):
+        client = _Client(self.PROJECT)
+        name = 'python_logger'
+
+        transport, _ = self._make_one(client, name)
+
+        transport.flush()
+
+        transport.worker.flush.assert_called()
+
 
 class Test_Worker(unittest.TestCase):
     NAME = 'python_logger'
@@ -233,6 +244,14 @@ def test__thread_main_batches(self):
         self.assertFalse(worker._cloud_logger._batch.commit_called)
         self.assertEqual(worker._queue.qsize(), 0)
 
+    def test_flush(self):
+        worker = self._make_one(_Logger(self.NAME))
+        worker._queue = mock.Mock(spec=queue.Queue)
+
+        # Queue is empty, should not block.
+        worker.flush()
+        worker._queue.join.assert_called()
+
 
 class _Thread(object):
 

logging/tests/unit/handlers/transports/test_base.py

@@ -32,3 +32,7 @@ def test_send_is_abstract(self):
         target = self._make_one()
         with self.assertRaises(NotImplementedError):
             target.send(None, None)
+
+    def test_flush_is_abstract_and_optional(self):
+        target = self._make_one()
+        target.flush()