Allow turning on JSON Detection in StackDriver (#6293)

* Allow turning on JSON Detection in StackDriver

To enable JSON detection in StackDriver, `jsonPayload` should
have no fields other than `message`. This patch lets clients
opt-in to this behavior, by creating a custom Transport.

Fixes googleapis/google-cloud-python#5799

Author: yuvipanda

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

@@ -92,12 +92,18 @@ class _Worker(object):
         than the grace_period. This means this is effectively the longest
         amount of time the background thread will hold onto log entries
         before sending them to the server.
+
+    :type includer_logger_name: bool
+    :param include_logger_name: (optional) Include python_logger field in
+        jsonPayload. Turn this off to enable json detection in log messages.
     """
 
     def __init__(self, cloud_logger, grace_period=_DEFAULT_GRACE_PERIOD,
                  max_batch_size=_DEFAULT_MAX_BATCH_SIZE,
-                 max_latency=_DEFAULT_MAX_LATENCY):
+                 max_latency=_DEFAULT_MAX_LATENCY,
+                 include_logger_name=True):
         self._cloud_logger = cloud_logger
+        self._include_logger_name = include_logger_name
         self._grace_period = grace_period
         self._max_batch_size = max_batch_size
         self._max_latency = max_latency
@@ -253,17 +259,21 @@ def enqueue(self, record, message, resource=None, labels=None,
         :param span_id: (optional) span_id within the trace for the log entry.
                         Specify the trace parameter if span_id is set.
         """
-        self._queue.put_nowait({
+
+        log_record = {
             'info': {
                 'message': message,
-                'python_logger': record.name,
             },
             'severity': record.levelname,
             'resource': resource,
             'labels': labels,
             'trace': trace,
             'span_id': span_id,
-        })
+        }
+
+        if self._include_logger_name:
+            log_record['info']['python_logger'] = record.name
+        self._queue.put_nowait(log_record)
 
     def flush(self):
         """Submit any pending log records."""
@@ -293,17 +303,24 @@ class BackgroundThreadTransport(Transport):
         than the grace_period. This means this is effectively the longest
         amount of time the background thread will hold onto log entries
         before sending them to the server.
+
+    :type includer_logger_name: bool
+    :param include_logger_name: (optional) Include python_logger field in
+                                jsonPayload. Turn this off to enable jso
+                                detection in log messages.
     """
 
     def __init__(self, client, name, grace_period=_DEFAULT_GRACE_PERIOD,
                  batch_size=_DEFAULT_MAX_BATCH_SIZE,
-                 max_latency=_DEFAULT_MAX_LATENCY):
+                 max_latency=_DEFAULT_MAX_LATENCY,
+                 include_logger_name=True):
         self.client = client
         logger = self.client.logger(name)
         self.worker = _Worker(logger,
                               grace_period=grace_period,
                               max_batch_size=batch_size,
-                              max_latency=max_latency)
+                              max_latency=max_latency,
+                              include_logger_name=include_logger_name)
         self.worker.start()
 
     def send(self, record, message, resource=None, labels=None,

tests/unit/handlers/transports/test_background_thread.py

@@ -175,6 +175,7 @@ def test_constructor(self):
         self.assertEqual(worker._grace_period, grace_period)
         self.assertEqual(worker._max_batch_size, max_batch_size)
         self.assertEqual(worker._max_latency, max_latency)
+        self.assertTrue(worker._include_logger_name)
         self.assertFalse(worker.is_alive)
         self.assertIsNone(worker._thread)
 
@@ -282,6 +283,23 @@ def test__thread_main(self):
         self.assertEqual(worker._cloud_logger._batch.commit_count, 2)
         self.assertEqual(worker._queue.qsize(), 0)
 
+    def test__thread_main_no_python_logger(self):
+        from google.cloud.logging.handlers.transports import background_thread
+
+        worker = self._make_one(_Logger(self.NAME), include_logger_name=False)
+        self.assertFalse(worker._include_logger_name)
+
+        # Enqueue one record and the termination signal.
+        self._enqueue_record(worker, '1')
+        worker._queue.put_nowait(background_thread._WORKER_TERMINATOR)
+
+        worker._thread_main()
+
+        self.assertEqual(len(worker._cloud_logger._batch.all_entries), 1)
+        self.assertFalse(
+            'python_logger' in worker._cloud_logger._batch.all_entries[0]
+        )
+
     def test__thread_main_error(self):
         from google.cloud.logging.handlers.transports import background_thread
 
@@ -421,9 +439,12 @@ def join(self, timeout=None):
 class _Batch(object):
 
     def __init__(self):
+        # Entries waiting to be committed
         self.entries = []
         self.commit_called = False
         self.commit_count = None
+        # All entries ever committed via this _Batch
+        self.all_entries = []
 
     def log_struct(
             self, info, severity=logging.INFO, resource=None, labels=None,
@@ -436,6 +457,7 @@ def log_struct(
         self.log_struct_called_with = (info, severity, resource, labels,
                                        trace, span_id)
         self.entries.append(info)
+        self.all_entries.append(info)
 
     def commit(self):
         self.commit_called = True