diff --git a/docs/index.rst b/docs/index.rst
index 595fd4703b8d..dfd557e17015 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -111,6 +111,8 @@
   logging-sink
   logging-stdlib-usage
   logging-handlers
+  logging-handlers-app-engine
+  logging-handlers-container-engine
   logging-transports-sync
   logging-transports-thread
   logging-transports-base
diff --git a/docs/logging-handlers-app-engine.rst b/docs/logging-handlers-app-engine.rst
new file mode 100644
index 000000000000..71c45e3690be
--- /dev/null
+++ b/docs/logging-handlers-app-engine.rst
@@ -0,0 +1,6 @@
+Google App Engine flexible Log Handler
+======================================
+
+.. automodule:: google.cloud.logging.handlers.app_engine
+  :members:
+  :show-inheritance:
diff --git a/docs/logging-handlers-container-engine.rst b/docs/logging-handlers-container-engine.rst
new file mode 100644
index 000000000000..a0c6b2bc9228
--- /dev/null
+++ b/docs/logging-handlers-container-engine.rst
@@ -0,0 +1,6 @@
+Google Container Engine Log Handler
+===================================
+
+.. automodule:: google.cloud.logging.handlers.container_engine
+  :members:
+  :show-inheritance:
diff --git a/docs/logging-usage.rst b/docs/logging-usage.rst
index 91a0f7ec63d8..e25b1645f75a 100644
--- a/docs/logging-usage.rst
+++ b/docs/logging-usage.rst
@@ -267,3 +267,108 @@ Delete a sink:
     :start-after: [START sink_delete]
     :end-before: [END sink_delete]
     :dedent: 4
+
+Integration with Python logging module
+--------------------------------------
+
+It's possible to tie the Python :mod:`logging` module directly into Google
+Stackdriver Logging. There are different handler options to accomplish this.
+To automatically pick the default for your current environment, use
+:meth:`~google.cloud.logging.client.Client.get_default_handler`.
+
+.. literalinclude:: logging_snippets.py
+    :start-after: [START create_default_handler]
+    :end-before: [END create_default_handler]
+    :dedent: 4
+
+It is also possible to attach the handler to the root Python logger, so that
+for example a plain ``logging.warn`` call would be sent to Stackdriver Logging,
+as well as any other loggers created. A helper method
+:meth:`~google.cloud.logging.client.Client.setup_logging` is provided
+to configure this automatically.
+
+.. literalinclude:: logging_snippets.py
+    :start-after: [START setup_logging]
+    :end-before: [END setup_logging]
+    :dedent: 4
+
+.. note::
+
+    To reduce cost and quota usage, do not enable Stackdriver logging
+    handlers while testing locally.
+
+You can also exclude certain loggers:
+
+.. literalinclude:: logging_snippets.py
+    :start-after: [START setup_logging_excludes]
+    :end-before: [END setup_logging_excludes]
+    :dedent: 4
+
+Cloud Logging Handler
+=====================
+
+If you prefer not to use
+:meth:`~google.cloud.logging.client.Client.get_default_handler`, you can
+directly create a
+:class:`~google.cloud.logging.handlers.handlers.CloudLoggingHandler` instance
+which will write directly to the API.
+
+.. literalinclude:: logging_snippets.py
+    :start-after: [START create_cloud_handler]
+    :end-before: [END create_cloud_handler]
+    :dedent: 4
+
+.. note::
+
+    This handler by default uses an asynchronous transport that sends log
+    entries on a background thread. However, the API call will still be made
+    in the same process. For other transport options, see the transports
+    section.
+
+All logs will go to a single custom log, which defaults to "python". The name
+of the Python logger will be included in the structured log entry under the
+"python_logger" field. You can change it by providing a name to the handler:
+
+.. literalinclude:: logging_snippets.py
+    :start-after: [START create_named_handler]
+    :end-before: [END create_named_handler]
+    :dedent: 4
+
+fluentd logging handlers
+========================
+
+Besides :class:`~google.cloud.logging.handlers.handlers.CloudLoggingHandler`,
+which writes directly to the API, two other handlers are provided.
+:class:`~google.cloud.logging.handlers.app_engine.AppEngineHandler`, which is
+recommended when running on the Google App Engine Flexible vanilla runtimes
+(i.e. your app.yaml contains ``runtime: python``), and
+:class:`~google.cloud.logging.handlers.container_engine.ContainerEngineHandler`
+, which is recommended when running on `Google Container Engine`_ with the
+Stackdriver Logging plugin enabled.
+
+:meth:`~google.cloud.logging.client.Client.get_default_handler` and
+:meth:`~google.cloud.logging.client.Client.setup_logging` will attempt to use
+the environment to automatically detect whether the code is running in
+these platforms and use the appropriate handler.
+
+In both cases, the fluentd agent is configured to automatically parse log files
+in an expected format and forward them to Stackdriver logging. The handlers
+provided help set the correct metadata such as log level so that logs can be
+filtered accordingly.
+
+Cloud Logging Handler transports
+=================================
+
+The :class:`~google.cloud.logging.handlers.handlers.CloudLoggingHandler`
+logging handler can use different transports. The default is
+:class:`~google.cloud.logging.handlers.BackgroundThreadTransport`.
+
+ 1. :class:`~google.cloud.logging.handlers.BackgroundThreadTransport` this is
+    the default. It writes entries on a background
+    :class:`python.threading.Thread`.
+
+ 1. :class:`~google.cloud.logging.handlers.SyncTransport` this handler does a
+    direct API call on each logging statement to write the entry.
+
+
+.. _Google Container Engine: https://cloud.google.com/container-engine/
diff --git a/docs/logging_snippets.py b/docs/logging_snippets.py
index cbef353db778..5fa539158f3c 100644
--- a/docs/logging_snippets.py
+++ b/docs/logging_snippets.py
@@ -324,6 +324,44 @@ def sink_pubsub(client, to_delete):
     to_delete.pop(0)
 
 
+@snippet
+def logging_handler(client):
+    # [START create_default_handler]
+    import logging
+    handler = client.get_default_handler()
+    cloud_logger = logging.getLogger('cloudLogger')
+    cloud_logger.setLevel(logging.INFO)
+    cloud_logger.addHandler(handler)
+    cloud_logger.error('bad news')
+    # [END create_default_handler]
+
+    # [START create_cloud_handler]
+    from google.cloud.logging.handlers import CloudLoggingHandler
+    handler = CloudLoggingHandler(client)
+    cloud_logger = logging.getLogger('cloudLogger')
+    cloud_logger.setLevel(logging.INFO)
+    cloud_logger.addHandler(handler)
+    cloud_logger.error('bad news')
+    # [END create_cloud_handler]
+
+    # [START create_named_handler]
+    handler = CloudLoggingHandler(client, name='mycustomlog')
+    # [END create_named_handler]
+
+
+@snippet
+def setup_logging(client):
+    import logging
+    # [START setup_logging]
+    client.setup_logging(log_level=logging.INFO)
+    # [END setup_logging]
+
+    # [START setup_logging_excludes]
+    client.setup_logging(log_level=logging.INFO,
+                         excluded_loggers=('werkzeug',))
+    # [END setup_logging_excludes]
+
+
 def _line_no(func):
     return func.__code__.co_firstlineno
 
diff --git a/logging/google/cloud/logging/client.py b/logging/google/cloud/logging/client.py
index b84fc9c6a736..c92f177eaac6 100644
--- a/logging/google/cloud/logging/client.py
+++ b/logging/google/cloud/logging/client.py
@@ -14,6 +14,7 @@
 
 """Client for interacting with the Google Stackdriver Logging API."""
 
+import logging
 import os
 
 try:
@@ -34,6 +35,12 @@
 from google.cloud.logging._http import _LoggingAPI as JSONLoggingAPI
 from google.cloud.logging._http import _MetricsAPI as JSONMetricsAPI
 from google.cloud.logging._http import _SinksAPI as JSONSinksAPI
+from google.cloud.logging.handlers import CloudLoggingHandler
+from google.cloud.logging.handlers import AppEngineHandler
+from google.cloud.logging.handlers import ContainerEngineHandler
+from google.cloud.logging.handlers import setup_logging
+from google.cloud.logging.handlers.handlers import EXCLUDED_LOGGER_DEFAULTS
+
 from google.cloud.logging.logger import Logger
 from google.cloud.logging.metric import Metric
 from google.cloud.logging.sink import Sink
@@ -42,6 +49,15 @@
 _DISABLE_GAX = os.getenv(DISABLE_GRPC, False)
 _USE_GAX = _HAVE_GAX and not _DISABLE_GAX
 
+_APPENGINE_FLEXIBLE_ENV_VM = 'GAE_APPENGINE_HOSTNAME'
+"""Environment variable set in App Engine when vm:true is set."""
+
+_APPENGINE_FLEXIBLE_ENV_FLEX = 'GAE_INSTANCE'
+"""Environment variable set in App Engine when env:flex is set."""
+
+_CONTAINER_ENGINE_ENV = 'KUBERNETES_SERVICE'
+"""Environment variable set in a Google Container Engine environment."""
+
 
 class Client(JSONClient):
     """Client to bundle configuration needed for API requests.
@@ -264,3 +280,40 @@ def list_metrics(self, page_size=None, page_token=None):
         """
         return self.metrics_api.list_metrics(
             self.project, page_size, page_token)
+
+    def get_default_handler(self):
+        """Return the default logging handler based on the local environment.
+
+        :rtype: :class:`logging.Handler`
+        :returns: The default log handler based on the environment
+        """
+        if (_APPENGINE_FLEXIBLE_ENV_VM in os.environ or
+                _APPENGINE_FLEXIBLE_ENV_FLEX in os.environ):
+            return AppEngineHandler()
+        elif _CONTAINER_ENGINE_ENV in os.environ:
+            return ContainerEngineHandler()
+        else:
+            return CloudLoggingHandler(self)
+
+    def setup_logging(self, log_level=logging.INFO,
+                      excluded_loggers=EXCLUDED_LOGGER_DEFAULTS):
+        """Attach default Stackdriver logging handler to the root logger.
+
+        This method uses the default log handler, obtained by
+        :meth:`~get_default_handler`, and attaches it to the root Python
+        logger, so that a call such as ``logging.warn``, as well as all child
+        loggers, will report to Stackdriver logging.
+
+        :type log_level: int
+        :param log_level: (Optional) Python logging log level. Defaults to
+                          :const:`logging.INFO`.
+
+        :type excluded_loggers: tuple
+        :param excluded_loggers: (Optional) The loggers to not attach the
+                                 handler to. This will always include the
+                                 loggers in the path of the logging client
+                                 itself.
+        """
+        handler = self.get_default_handler()
+        setup_logging(handler, log_level=log_level,
+                      excluded_loggers=excluded_loggers)
diff --git a/logging/google/cloud/logging/handlers/__init__.py b/logging/google/cloud/logging/handlers/__init__.py
index 57d08af8637f..9745296e9782 100644
--- a/logging/google/cloud/logging/handlers/__init__.py
+++ b/logging/google/cloud/logging/handlers/__init__.py
@@ -14,5 +14,8 @@
 
 """Python :mod:`logging` handlers for Google Cloud Logging."""
 
+from google.cloud.logging.handlers.app_engine import AppEngineHandler
+from google.cloud.logging.handlers.container_engine import (
+    ContainerEngineHandler)
 from google.cloud.logging.handlers.handlers import CloudLoggingHandler
 from google.cloud.logging.handlers.handlers import setup_logging
diff --git a/logging/google/cloud/logging/handlers/_helpers.py b/logging/google/cloud/logging/handlers/_helpers.py
new file mode 100644
index 000000000000..81adcf0eb545
--- /dev/null
+++ b/logging/google/cloud/logging/handlers/_helpers.py
@@ -0,0 +1,39 @@
+# Copyright 2016 Google Inc. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Helper functions for logging handlers."""
+
+import math
+import json
+
+
+def format_stackdriver_json(record, message):
+    """Helper to format a LogRecord in in Stackdriver fluentd format.
+
+        :rtype: str
+        :returns: JSON str to be written to the log file.
+    """
+    subsecond, second = math.modf(record.created)
+
+    payload = {
+        'message': message,
+        'timestamp': {
+            'seconds': int(second),
+            'nanos': int(subsecond * 1e9),
+        },
+        'thread': record.thread,
+        'severity': record.levelname,
+    }
+
+    return json.dumps(payload)
diff --git a/logging/google/cloud/logging/handlers/app_engine.py b/logging/google/cloud/logging/handlers/app_engine.py
new file mode 100644
index 000000000000..4184c2054b1a
--- /dev/null
+++ b/logging/google/cloud/logging/handlers/app_engine.py
@@ -0,0 +1,73 @@
+# Copyright 2016 Google Inc. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Logging handler for App Engine Flexible
+
+Logs to the well-known file that the fluentd sidecar container on App Engine
+Flexible is configured to read from and send to Stackdriver Logging.
+
+See the fluentd configuration here:
+
+https://github.com/GoogleCloudPlatform/appengine-sidecars-docker/tree/master/fluentd_logger
+"""
+
+# This file is largely copied from:
+#  https://github.com/GoogleCloudPlatform/python-compat-runtime/blob/master
+# /appengine-vmruntime/vmruntime/cloud_logging.py
+
+import logging.handlers
+import os
+
+from google.cloud.logging.handlers._helpers import format_stackdriver_json
+
+_LOG_PATH_TEMPLATE = '/var/log/app_engine/app.{pid}.json'
+_MAX_LOG_BYTES = 128 * 1024 * 1024
+_LOG_FILE_COUNT = 3
+
+
+class AppEngineHandler(logging.handlers.RotatingFileHandler):
+    """A handler that writes to the App Engine fluentd Stackdriver log file.
+
+    Writes to the file that the fluentd agent on App Engine Flexible is
+    configured to discover logs and send them to  Stackdriver Logging.
+    Log entries are wrapped in JSON and with appropriate metadata. The
+    process of converting the user's formatted logs into a JSON payload for
+    Stackdriver Logging consumption is implemented as part of the handler
+    itself, and not as a formatting step, so as not to interfere with
+    user-defined logging formats.
+    """
+
+    def __init__(self):
+        """Construct the handler
+
+        Large log entries will get mangled if multiple workers write to the
+        same file simultaneously, so we'll use the worker's PID to pick a log
+        filename.
+        """
+        self.filename = _LOG_PATH_TEMPLATE.format(pid=os.getpid())
+        super(AppEngineHandler, self).__init__(self.filename,
+                                               maxBytes=_MAX_LOG_BYTES,
+                                               backupCount=_LOG_FILE_COUNT)
+
+    def format(self, record):
+        """Format the specified record into the expected JSON structure.
+
+        :type record: :class:`~logging.LogRecord`
+        :param record: the log record
+
+        :rtype: str
+        :returns: JSON str to be written to the log file
+        """
+        message = super(AppEngineHandler, self).format(record)
+        return format_stackdriver_json(record, message)
diff --git a/logging/google/cloud/logging/handlers/container_engine.py b/logging/google/cloud/logging/handlers/container_engine.py
new file mode 100644
index 000000000000..8beb7d076a4b
--- /dev/null
+++ b/logging/google/cloud/logging/handlers/container_engine.py
@@ -0,0 +1,44 @@
+# Copyright 2016 Google Inc. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Logging handler for Google Container Engine (GKE).
+
+Formats log messages in a JSON format, so that Kubernetes clusters with the
+fluentd Google Cloud plugin installed can format their log messages so that
+metadata such as log level is properly captured.
+"""
+
+import logging.handlers
+
+from google.cloud.logging.handlers._helpers import format_stackdriver_json
+
+
+class ContainerEngineHandler(logging.StreamHandler):
+    """Handler to format log messages the format expected by GKE fluent.
+
+    This handler is written to format messages for the Google Container Engine
+    (GKE) fluentd plugin, so that metadata such as log level are properly set.
+    """
+
+    def format(self, record):
+        """Format the message into JSON expected by fluentd.
+
+        :type record: :class:`~logging.LogRecord`
+        :param record: the log record
+
+        :rtype: str
+        :returns: A JSON string formatted for GKE fluentd.
+        """
+        message = super(ContainerEngineHandler, self).format(record)
+        return format_stackdriver_json(record, message)
diff --git a/logging/google/cloud/logging/handlers/handlers.py b/logging/google/cloud/logging/handlers/handlers.py
index e3b6d5b30da4..4cf3f0cb20e9 100644
--- a/logging/google/cloud/logging/handlers/handlers.py
+++ b/logging/google/cloud/logging/handlers/handlers.py
@@ -12,30 +12,25 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""Python :mod:`logging` handlers for Google Cloud Logging."""
+"""Python :mod:`logging` handlers for Stackdriver Logging."""
 
 import logging
 
 from google.cloud.logging.handlers.transports import BackgroundThreadTransport
 
-
-EXCLUDE_LOGGER_DEFAULTS = (
-    'google.cloud',
-    'oauth2client'
-)
-
 DEFAULT_LOGGER_NAME = 'python'
 
+EXCLUDED_LOGGER_DEFAULTS = ('google.cloud', 'oauth2client')
+
 
 class CloudLoggingHandler(logging.StreamHandler):
-    """Python standard ``logging`` handler.
+    """Handler that directly makes Stackdriver logging API calls.
 
-    This handler can be used to route Python standard logging messages
-    directly to the Stackdriver Logging API.
+    This is a Python standard ``logging`` handler using that can be used to
+    route Python standard logging messages directly to the Stackdriver
+    Logging API.
 
-    Note that this handler currently only supports a synchronous API call,
-    which means each logging statement that uses this handler will require
-    an API call.
+    This handler supports both an asynchronous and synchronous transport.
 
     :type client: :class:`google.cloud.logging.client`
     :param client: the authenticated Google Cloud Logging client for this
@@ -93,8 +88,9 @@ def emit(self, record):
         self.transport.send(record, message)
 
 
-def setup_logging(handler, excluded_loggers=EXCLUDE_LOGGER_DEFAULTS):
-    """Attach the ``CloudLogging`` handler to the Python root logger
+def setup_logging(handler, excluded_loggers=EXCLUDED_LOGGER_DEFAULTS,
+                  log_level=logging.INFO):
+    """Attach a logging handler to the Python root logger
 
     Excludes loggers that this library itself uses to avoid
     infinite recursion.
@@ -103,9 +99,13 @@ def setup_logging(handler, excluded_loggers=EXCLUDE_LOGGER_DEFAULTS):
     :param handler: the handler to attach to the global handler
 
     :type excluded_loggers: tuple
-    :param excluded_loggers: The loggers to not attach the handler to. This
-                             will always include the loggers in the path of
-                             the logging client itself.
+    :param excluded_loggers: (Optional) The loggers to not attach the handler
+                             to. This will always include the loggers in the
+                             path of the logging client itself.
+
+    :type log_level: int
+    :param log_level: (Optional) Python logging log level. Defaults to
+                      :const:`logging.INFO`.
 
     Example:
 
@@ -123,8 +123,9 @@ def setup_logging(handler, excluded_loggers=EXCLUDE_LOGGER_DEFAULTS):
         logging.error('bad news')  # API call
 
     """
-    all_excluded_loggers = set(excluded_loggers + EXCLUDE_LOGGER_DEFAULTS)
+    all_excluded_loggers = set(excluded_loggers + EXCLUDED_LOGGER_DEFAULTS)
     logger = logging.getLogger()
+    logger.setLevel(log_level)
     logger.addHandler(handler)
     logger.addHandler(logging.StreamHandler())
     for logger_name in all_excluded_loggers:
diff --git a/logging/google/cloud/logging/handlers/transports/background_thread.py b/logging/google/cloud/logging/handlers/transports/background_thread.py
index 144bccafc838..aa50e0d3ffc1 100644
--- a/logging/google/cloud/logging/handlers/transports/background_thread.py
+++ b/logging/google/cloud/logging/handlers/transports/background_thread.py
@@ -21,9 +21,10 @@
 import copy
 import threading
 
-from google.cloud.logging.client import Client
 from google.cloud.logging.handlers.transports.base import Transport
 
+_WORKER_THREAD_NAME = 'google.cloud.logging.handlers.transport.Worker'
+
 
 class _Worker(object):
     """A threaded worker that writes batches of log entries
@@ -96,8 +97,7 @@ def _start(self):
         try:
             self._entries_condition.acquire()
             self._thread = threading.Thread(
-                target=self._run,
-                name='google.cloud.logging.handlers.transport.Worker')
+                target=self._run, name=_WORKER_THREAD_NAME)
             self._thread.setDaemon(True)
             self._thread.start()
         finally:
@@ -152,9 +152,8 @@ class BackgroundThreadTransport(Transport):
     def __init__(self, client, name):
         http = copy.deepcopy(client._connection.http)
         http = client._connection.credentials.authorize(http)
-        self.client = Client(client.project,
-                             client._connection.credentials,
-                             http)
+        self.client = client.__class__(client.project,
+                                       client._connection.credentials, http)
         logger = self.client.logger(name)
         self.worker = _Worker(logger)
 
diff --git a/logging/unit_tests/handlers/test_app_engine.py b/logging/unit_tests/handlers/test_app_engine.py
new file mode 100644
index 000000000000..9be8a2bec9b3
--- /dev/null
+++ b/logging/unit_tests/handlers/test_app_engine.py
@@ -0,0 +1,57 @@
+# Copyright 2016 Google Inc. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+
+
+class TestAppEngineHandlerHandler(unittest.TestCase):
+    PROJECT = 'PROJECT'
+
+    def _get_target_class(self):
+        from google.cloud.logging.handlers.app_engine import AppEngineHandler
+
+        return AppEngineHandler
+
+    def _make_one(self, *args, **kw):
+        import tempfile
+
+        from google.cloud._testing import _Monkey
+        from google.cloud.logging.handlers import app_engine as _MUT
+
+        tmpdir = tempfile.mktemp()
+        with _Monkey(_MUT, _LOG_PATH_TEMPLATE=tmpdir):
+            return self._get_target_class()(*args, **kw)
+
+    def test_format(self):
+        import json
+        import logging
+
+        handler = self._make_one()
+        logname = 'loggername'
+        message = 'hello world'
+        record = logging.LogRecord(logname, logging.INFO, None,
+                                   None, message, None, None)
+        record.created = 5.03
+        expected_payload = {
+            'message': message,
+            'timestamp': {
+                'seconds': 5,
+                'nanos': int(.03 * 1e9),
+            },
+            'thread': record.thread,
+            'severity': record.levelname,
+        }
+        payload = handler.format(record)
+
+        self.assertEqual(payload, json.dumps(expected_payload))
diff --git a/logging/unit_tests/handlers/test_container_engine.py b/logging/unit_tests/handlers/test_container_engine.py
new file mode 100644
index 000000000000..b8ce0dc436f3
--- /dev/null
+++ b/logging/unit_tests/handlers/test_container_engine.py
@@ -0,0 +1,51 @@
+# Copyright 2016 Google Inc. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+
+
+class TestContainerEngineHandler(unittest.TestCase):
+    PROJECT = 'PROJECT'
+
+    def _get_target_class(self):
+        from google.cloud.logging.handlers.container_engine import (
+            ContainerEngineHandler)
+
+        return ContainerEngineHandler
+
+    def _make_one(self, *args, **kw):
+        return self._get_target_class()(*args, **kw)
+
+    def test_format(self):
+        import logging
+        import json
+
+        handler = self._make_one()
+        logname = 'loggername'
+        message = 'hello world'
+        record = logging.LogRecord(logname, logging.INFO, None, None,
+                                   message, None, None)
+        record.created = 5.03
+        expected_payload = {
+            'message': message,
+            'timestamp': {
+                'seconds': 5,
+                'nanos': int(.03 * 1e9)
+            },
+            'thread': record.thread,
+            'severity': record.levelname,
+        }
+        payload = handler.format(record)
+
+        self.assertEqual(payload, json.dumps(expected_payload))
diff --git a/logging/unit_tests/handlers/test_handlers.py b/logging/unit_tests/handlers/test_handlers.py
index 54c38f9b82cb..234b2991df45 100644
--- a/logging/unit_tests/handlers/test_handlers.py
+++ b/logging/unit_tests/handlers/test_handlers.py
@@ -36,12 +36,13 @@ def test_ctor(self):
     def test_emit(self):
         client = _Client(self.PROJECT)
         handler = self._make_one(client, transport=_Transport)
-        LOGNAME = 'loggername'
-        MESSAGE = 'hello world'
-        record = _Record(LOGNAME, logging.INFO, MESSAGE)
+        logname = 'loggername'
+        message = 'hello world'
+        record = logging.LogRecord(logname, logging, None, None, message,
+                                   None, None)
         handler.emit(record)
 
-        self.assertEqual(handler.transport.send_called_with, (record, MESSAGE))
+        self.assertEqual(handler.transport.send_called_with, (record, message))
 
 
 class TestSetupLogging(unittest.TestCase):
@@ -100,20 +101,6 @@ def __init__(self, project):
         self.project = project
 
 
-class _Record(object):
-
-    def __init__(self, name, level, message):
-        self.name = name
-        self.levelname = level
-        self.message = message
-        self.exc_info = None
-        self.exc_text = None
-        self.stack_info = None
-
-    def getMessage(self):
-        return self.message
-
-
 class _Transport(object):
 
     def __init__(self, client, name):
diff --git a/logging/unit_tests/handlers/transports/test_background_thread.py b/logging/unit_tests/handlers/transports/test_background_thread.py
index 3695c591288c..eb9204b4e2ae 100644
--- a/logging/unit_tests/handlers/transports/test_background_thread.py
+++ b/logging/unit_tests/handlers/transports/test_background_thread.py
@@ -42,16 +42,17 @@ def test_send(self):
         transport = self._make_one(client, NAME)
         transport.worker.batch = client.logger(NAME).batch()
 
-        PYTHON_LOGGER_NAME = 'mylogger'
-        MESSAGE = 'hello world'
-        record = _Record(PYTHON_LOGGER_NAME, logging.INFO, MESSAGE)
-        transport.send(record, MESSAGE)
+        python_logger_name = 'mylogger'
+        message = 'hello world'
+        record = logging.LogRecord(python_logger_name, logging.INFO,
+                                   None, None, message, None, None)
+        transport.send(record, message)
 
         EXPECTED_STRUCT = {
-            'message': MESSAGE,
-            'python_logger': PYTHON_LOGGER_NAME
+            'message': message,
+            'python_logger': python_logger_name
         }
-        EXPECTED_SENT = (EXPECTED_STRUCT, logging.INFO)
+        EXPECTED_SENT = (EXPECTED_STRUCT, 'INFO')
         self.assertEqual(transport.worker.batch.log_struct_called_with,
                          EXPECTED_SENT)
 
@@ -77,9 +78,11 @@ def test_run(self):
         logger = _Logger(NAME)
         worker = self._make_one(logger)
 
-        PYTHON_LOGGER_NAME = 'mylogger'
-        MESSAGE = 'hello world'
-        record = _Record(PYTHON_LOGGER_NAME, logging.INFO, MESSAGE)
+        python_logger_name = 'mylogger'
+        message = 'hello world'
+        record = logging.LogRecord(python_logger_name,
+                                   logging.INFO, None, None,
+                                   message, None, None)
 
         worker._start()
 
@@ -91,7 +94,7 @@ def test_run(self):
         while not worker.started:
             time.sleep(1)  # pragma: NO COVER
 
-        worker.enqueue(record, MESSAGE)
+        worker.enqueue(record, message)
         # Set timeout to none so worker thread finishes
         worker._stop_timeout = None
         worker._stop()
@@ -99,20 +102,22 @@ def test_run(self):
 
     def test_run_after_stopped(self):
         # No-op
-        NAME = 'python_logger'
-        logger = _Logger(NAME)
+        name = 'python_logger'
+        logger = _Logger(name)
         worker = self._make_one(logger)
 
-        PYTHON_LOGGER_NAME = 'mylogger'
-        MESSAGE = 'hello world'
-        record = _Record(PYTHON_LOGGER_NAME, logging.INFO, MESSAGE)
+        python_logger_name = 'mylogger'
+        message = 'hello world'
+        record = logging.LogRecord(python_logger_name,
+                                   logging.INFO, None, None,
+                                   message, None, None)
 
         worker._start()
         while not worker.started:
             time.sleep(1)  # pragma: NO COVER
         worker._stop_timeout = None
         worker._stop()
-        worker.enqueue(record, MESSAGE)
+        worker.enqueue(record, message)
         self.assertFalse(worker.batch.commit_called)
         worker._stop()
 
@@ -122,11 +127,13 @@ def test_run_enqueue_early(self):
         logger = _Logger(NAME)
         worker = self._make_one(logger)
 
-        PYTHON_LOGGER_NAME = 'mylogger'
-        MESSAGE = 'hello world'
-        record = _Record(PYTHON_LOGGER_NAME, logging.INFO, MESSAGE)
+        python_logger_name = 'mylogger'
+        message = 'hello world'
+        record = logging.LogRecord(python_logger_name,
+                                   logging.INFO, None, None,
+                                   message, None, None)
 
-        worker.enqueue(record, MESSAGE)
+        worker.enqueue(record, message)
         worker._start()
         while not worker.started:
             time.sleep(1)  # pragma: NO COVER
@@ -135,17 +142,6 @@ def test_run_enqueue_early(self):
         self.assertTrue(worker.stopped)
 
 
-class _Record(object):
-
-    def __init__(self, name, level, message):
-        self.name = name
-        self.levelname = level
-        self.message = message
-        self.exc_info = None
-        self.exc_text = None
-        self.stack_info = None
-
-
 class _Batch(object):
 
     def __init__(self):
@@ -186,8 +182,10 @@ def batch(self):
 
 class _Client(object):
 
-    def __init__(self, project):
+    def __init__(self, project, http=None, credentials=None):
         self.project = project
+        self.http = http
+        self.credentials = credentials
         self._connection = _Connection()
 
     def logger(self, name):  # pylint: disable=unused-argument
diff --git a/logging/unit_tests/handlers/transports/test_sync.py b/logging/unit_tests/handlers/transports/test_sync.py
index 54e14dcbdfff..6650eb8a9d2e 100644
--- a/logging/unit_tests/handlers/transports/test_sync.py
+++ b/logging/unit_tests/handlers/transports/test_sync.py
@@ -36,33 +36,24 @@ def test_ctor(self):
 
     def test_send(self):
         client = _Client(self.PROJECT)
-        STACKDRIVER_LOGGER_NAME = 'python'
-        PYTHON_LOGGER_NAME = 'mylogger'
-        transport = self._make_one(client, STACKDRIVER_LOGGER_NAME)
-        MESSAGE = 'hello world'
-        record = _Record(PYTHON_LOGGER_NAME, logging.INFO, MESSAGE)
 
-        transport.send(record, MESSAGE)
+        stackdriver_logger_name = 'python'
+        python_logger_name = 'mylogger'
+        transport = self._make_one(client, stackdriver_logger_name)
+        message = 'hello world'
+        record = logging.LogRecord(python_logger_name, logging.INFO,
+                                   None, None, message, None, None)
+
+        transport.send(record, message)
         EXPECTED_STRUCT = {
-            'message': MESSAGE,
-            'python_logger': PYTHON_LOGGER_NAME
+            'message': message,
+            'python_logger': python_logger_name,
         }
-        EXPECTED_SENT = (EXPECTED_STRUCT, logging.INFO)
+        EXPECTED_SENT = (EXPECTED_STRUCT, 'INFO')
         self.assertEqual(
             transport.logger.log_struct_called_with, EXPECTED_SENT)
 
 
-class _Record(object):
-
-    def __init__(self, name, level, message):
-        self.name = name
-        self.levelname = level
-        self.message = message
-        self.exc_info = None
-        self.exc_text = None
-        self.stack_info = None
-
-
 class _Logger(object):
 
     def __init__(self, name):
diff --git a/logging/unit_tests/test_client.py b/logging/unit_tests/test_client.py
index 7e5173932ff1..6e7fc8f80f56 100644
--- a/logging/unit_tests/test_client.py
+++ b/logging/unit_tests/test_client.py
@@ -549,6 +549,79 @@ def test_list_metrics_with_paging(self):
             },
         })
 
+    def test_get_default_handler_app_engine(self):
+        import os
+        from google.cloud._testing import _Monkey
+        from google.cloud.logging.client import _APPENGINE_FLEXIBLE_ENV_VM
+        from google.cloud.logging.handlers import app_engine as _MUT
+        from google.cloud.logging.handlers import AppEngineHandler
+
+        client = self._make_one(project=self.PROJECT,
+                                credentials=_Credentials(),
+                                use_gax=False)
+
+        with _Monkey(_MUT, _LOG_PATH_TEMPLATE='{pid}'):
+            with _Monkey(os, environ={_APPENGINE_FLEXIBLE_ENV_VM: 'True'}):
+                handler = client.get_default_handler()
+
+        self.assertIsInstance(handler, AppEngineHandler)
+
+    def test_get_default_handler_container_engine(self):
+        import os
+        from google.cloud._testing import _Monkey
+        from google.cloud.logging.client import _CONTAINER_ENGINE_ENV
+        from google.cloud.logging.handlers import ContainerEngineHandler
+
+        client = self._make_one(project=self.PROJECT,
+                                credentials=_Credentials(),
+                                use_gax=False)
+
+        with _Monkey(os, environ={_CONTAINER_ENGINE_ENV: 'True'}):
+            handler = client.get_default_handler()
+
+        self.assertIsInstance(handler, ContainerEngineHandler)
+
+    def test_get_default_handler_general(self):
+        import httplib2
+        import mock
+        from google.cloud.logging.handlers import CloudLoggingHandler
+
+        http_mock = mock.Mock(spec=httplib2.Http)
+        credentials = _Credentials()
+        deepcopy = mock.Mock(return_value=http_mock)
+
+        with mock.patch('copy.deepcopy', new=deepcopy):
+            client = self._make_one(project=self.PROJECT,
+                                    credentials=credentials,
+                                    use_gax=False)
+            handler = client.get_default_handler()
+            deepcopy.assert_called_once_with(client._connection.http)
+
+        self.assertIsInstance(handler, CloudLoggingHandler)
+        self.assertTrue(credentials.authorized, http_mock)
+
+    def test_setup_logging(self):
+        import httplib2
+        import mock
+
+        http_mock = mock.Mock(spec=httplib2.Http)
+        deepcopy = mock.Mock(return_value=http_mock)
+        setup_logging = mock.Mock()
+
+        credentials = _Credentials()
+
+        with mock.patch('copy.deepcopy', new=deepcopy):
+            with mock.patch('google.cloud.logging.client.setup_logging',
+                            new=setup_logging):
+                client = self._make_one(project=self.PROJECT,
+                                        credentials=credentials,
+                                        use_gax=False)
+                client.setup_logging()
+                deepcopy.assert_called_once_with(client._connection.http)
+
+        setup_logging.assert_called()
+        self.assertTrue(credentials.authorized, http_mock)
+
 
 class _Credentials(object):
 
@@ -562,6 +635,9 @@ def create_scoped(self, scope):
         self._scopes = scope
         return self
 
+    def authorize(self, http):
+        self.authorized = http
+
 
 class _Connection(object):