diff --git a/docs/index.rst b/docs/index.rst
index 5475d3447637a..99215f213e6c4 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -107,6 +107,9 @@
   logging-metric
   logging-sink
   logging-handlers
+  logging-transports-sync
+  logging-transports-thread
+  logging-transports-base
 
 .. toctree::
   :maxdepth: 0
diff --git a/docs/logging-handlers.rst b/docs/logging-handlers.rst
index f4fa67323d8c4..c612d0f1129b2 100644
--- a/docs/logging-handlers.rst
+++ b/docs/logging-handlers.rst
@@ -1,6 +1,6 @@
 Python Logging Module Handler
 ==============================
 
-.. automodule:: gcloud.logging.handlers
+.. automodule:: gcloud.logging.handlers.handlers
   :members:
   :show-inheritance:
diff --git a/docs/logging-transports-base.rst b/docs/logging-transports-base.rst
new file mode 100644
index 0000000000000..b01b55f1b7323
--- /dev/null
+++ b/docs/logging-transports-base.rst
@@ -0,0 +1,6 @@
+Python Logging Handler Sync Transport
+======================================
+
+.. automodule:: gcloud.logging.handlers.transports.base
+  :members:
+  :show-inheritance:
diff --git a/docs/logging-transports-sync.rst b/docs/logging-transports-sync.rst
new file mode 100644
index 0000000000000..88f2cf172c199
--- /dev/null
+++ b/docs/logging-transports-sync.rst
@@ -0,0 +1,6 @@
+Python Logging Handler Sync Transport
+======================================
+
+.. automodule:: gcloud.logging.handlers.transports.sync
+  :members:
+  :show-inheritance:
diff --git a/docs/logging-transports-thread.rst b/docs/logging-transports-thread.rst
new file mode 100644
index 0000000000000..97f41730a3a29
--- /dev/null
+++ b/docs/logging-transports-thread.rst
@@ -0,0 +1,7 @@
+Python Logging Handler Threaded Transport
+=========================================
+
+
+.. automodule:: gcloud.logging.handlers.transports.background_thread
+  :members:
+  :show-inheritance:
diff --git a/docs/logging-usage.rst b/docs/logging-usage.rst
index ae5156f5f58ac..1100227a078bf 100644
--- a/docs/logging-usage.rst
+++ b/docs/logging-usage.rst
@@ -402,12 +402,21 @@ Logging client.
     >>> cloud_logger = logging.getLogger('cloudLogger')
     >>> cloud_logger.setLevel(logging.INFO) # defaults to WARN
     >>> cloud_logger.addHandler(handler)
-    >>> cloud_logger.error('bad news') # API call
+    >>> cloud_logger.error('bad news')
 
 .. note::
 
-    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 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:
+
+.. doctest::
+
+    >>> handler = CloudLoggingHandler(client, name="mycustomlog")
 
 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 Cloud Logging, as well as any other loggers created. However,
@@ -424,4 +433,24 @@ this automatically:
     >>> handler = CloudLoggingHandler(client)
     >>> logging.getLogger().setLevel(logging.INFO) # defaults to WARN
     >>> setup_logging(handler)
-    >>> logging.error('bad news') # API call
+    >>> logging.error('bad news')
+
+You can also exclude certain loggers:
+
+.. doctest::
+
+   >>> setup_logging(handler, excluded_loggers=('werkzeug',)))
+
+
+
+Python logging handler transports
+==================================
+
+The Python logging handler can use different transports. The default is
+:class:`gcloud.logging.handlers.BackgroundThreadTransport`.
+
+ 1. :class:`gcloud.logging.handlers.BackgroundThreadTransport` this is the default. It writes
+ entries on a background :class:`python.threading.Thread`.
+
+ 1. :class:`gcloud.logging.handlers.SyncTransport` this handler does a direct API call on each
+ logging statement to write the entry.
diff --git a/gcloud/logging/handlers/__init__.py b/gcloud/logging/handlers/__init__.py
new file mode 100644
index 0000000000000..2f12851ba6e09
--- /dev/null
+++ b/gcloud/logging/handlers/__init__.py
@@ -0,0 +1,18 @@
+# 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.
+
+"""Python :mod:`logging` handlers for Google Cloud Logging."""
+
+from gcloud.logging.handlers.handlers import CloudLoggingHandler
+from gcloud.logging.handlers.handlers import setup_logging
diff --git a/gcloud/logging/handlers.py b/gcloud/logging/handlers/handlers.py
similarity index 61%
rename from gcloud/logging/handlers.py
rename to gcloud/logging/handlers/handlers.py
index 9165b091f5db0..8c12072651dca 100644
--- a/gcloud/logging/handlers.py
+++ b/gcloud/logging/handlers/handlers.py
@@ -16,18 +16,22 @@
 
 import logging
 
+from gcloud.logging.handlers.transports import BackgroundThreadTransport
+
+
 EXCLUDE_LOGGER_DEFAULTS = (
     'gcloud',
-    'oauth2client.client'
+    'oauth2client'
 )
 
+DEFAULT_LOGGER_NAME = 'python'
+
 
-class CloudLoggingHandler(logging.StreamHandler, object):
-    """Python standard logging handler to log messages to the Google Cloud
-    Logging API.
+class CloudLoggingHandler(logging.StreamHandler):
+    """Python standard ``logging`` handler.
 
-    This handler can be used to route Python standard logging messages to
-    Google Cloud logging.
+    This handler can be used to route Python standard logging messages
+    directly to the Google Cloud Logging API.
 
     Note that this handler currently only supports a synchronous API call,
     which means each logging statement that uses this handler will require
@@ -37,6 +41,18 @@ class CloudLoggingHandler(logging.StreamHandler, object):
     :param client: the authenticated gcloud logging client for this handler
                    to use
 
+    :type name: str
+    :param name: the name of the custom log in Stackdriver Logging. Defaults
+                 to 'python'. The name of the Python logger will be represented
+                 in the ``python_logger`` field.
+
+    :type transport: type
+    :param transport: Class for creating new transport objects. It should
+                      extend from the base :class:`.Transport` type and
+                      implement :meth`.Transport.send`. Defaults to
+                      :class:`.BackgroundThreadTransport`. The other
+                      option is :class:`.SyncTransport`.
+
     Example:
 
     .. doctest::
@@ -51,30 +67,37 @@ class CloudLoggingHandler(logging.StreamHandler, object):
         cloud_logger.setLevel(logging.INFO)
         cloud_logger.addHandler(handler)
 
-        cloud.logger.error("bad news") # API call
+        cloud.logger.error('bad news')  # API call
 
     """
 
-    def __init__(self, client):
+    def __init__(self, client,
+                 name=DEFAULT_LOGGER_NAME,
+                 transport=BackgroundThreadTransport):
         super(CloudLoggingHandler, self).__init__()
+        self.name = name
         self.client = client
+        self.transport = transport(client, name)
 
     def emit(self, record):
-        """
-        Overrides the default emit behavior of StreamHandler.
+        """Actually log the specified logging record.
+
+        Overrides the default emit behavior of ``StreamHandler``.
 
         See: https://docs.python.org/2/library/logging.html#handler-objects
+
+        :type record: :class:`logging.LogRecord`
+        :param record: The record to be logged.
         """
         message = super(CloudLoggingHandler, self).format(record)
-        logger = self.client.logger(record.name)
-        logger.log_struct({"message": message},
-                          severity=record.levelname)
+        self.transport.send(record, message)
 
 
 def setup_logging(handler, excluded_loggers=EXCLUDE_LOGGER_DEFAULTS):
-    """Helper function to attach the CloudLoggingAPI handler to the Python
-    root logger, while excluding loggers this library itself uses to avoid
-    infinite recursion
+    """Attach the ``CloudLogging`` handler to the Python root logger
+
+    Excludes loggers that this library itself uses to avoid
+    infinite recursion.
 
     :type handler: :class:`logging.handler`
     :param handler: the handler to attach to the global handler
@@ -90,14 +113,14 @@ def setup_logging(handler, excluded_loggers=EXCLUDE_LOGGER_DEFAULTS):
 
         import logging
         import gcloud.logging
-        from gcloud.logging.handlers import CloudLoggingAPIHandler
+        from gcloud.logging.handlers import CloudLoggingHandler
 
         client = gcloud.logging.Client()
         handler = CloudLoggingHandler(client)
-        setup_logging(handler)
+        gcloud.logging.setup_logging(handler)
         logging.getLogger().setLevel(logging.DEBUG)
 
-        logging.error("bad news") # API call
+        logging.error('bad news')  # API call
 
     """
     all_excluded_loggers = set(excluded_loggers + EXCLUDE_LOGGER_DEFAULTS)
diff --git a/gcloud/logging/test_handlers.py b/gcloud/logging/handlers/test_handlers.py
similarity index 82%
rename from gcloud/logging/test_handlers.py
rename to gcloud/logging/handlers/test_handlers.py
index d525028cddb5b..60ede6a5bf6ca 100644
--- a/gcloud/logging/test_handlers.py
+++ b/gcloud/logging/handlers/test_handlers.py
@@ -1,4 +1,3 @@
-#!/usr/bin/env python
 # Copyright 2016 Google Inc. All Rights Reserved.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -22,7 +21,7 @@ class TestCloudLoggingHandler(unittest.TestCase):
     PROJECT = 'PROJECT'
 
     def _getTargetClass(self):
-        from gcloud.logging.handlers import CloudLoggingHandler
+        from gcloud.logging.handlers.handlers import CloudLoggingHandler
         return CloudLoggingHandler
 
     def _makeOne(self, *args, **kw):
@@ -30,24 +29,24 @@ def _makeOne(self, *args, **kw):
 
     def test_ctor(self):
         client = _Client(self.PROJECT)
-        handler = self._makeOne(client)
+        handler = self._makeOne(client, transport=_Transport)
         self.assertEqual(handler.client, client)
 
     def test_emit(self):
         client = _Client(self.PROJECT)
-        handler = self._makeOne(client)
+        handler = self._makeOne(client, transport=_Transport)
         LOGNAME = 'loggername'
         MESSAGE = 'hello world'
         record = _Record(LOGNAME, logging.INFO, MESSAGE)
         handler.emit(record)
-        self.assertEqual(client.logger(LOGNAME).log_struct_called_with,
-                         ({'message': MESSAGE}, logging.INFO))
+
+        self.assertEqual(handler.transport.send_called_with, (record, MESSAGE))
 
 
 class TestSetupLogging(unittest.TestCase):
 
     def _callFUT(self, handler, excludes=None):
-        from gcloud.logging.handlers import setup_logging
+        from gcloud.logging.handlers.handlers import setup_logging
         if excludes:
             return setup_logging(handler, excluded_loggers=excludes)
         else:
@@ -94,20 +93,10 @@ def release(self):
         pass  # pragma: NO COVER
 
 
-class _Logger(object):
-
-    def log_struct(self, message, severity=None):
-        self.log_struct_called_with = (message, severity)
-
-
 class _Client(object):
 
     def __init__(self, project):
         self.project = project
-        self.logger_ = _Logger()
-
-    def logger(self, _):  # pylint: disable=unused-argument
-        return self.logger_
 
 
 class _Record(object):
@@ -122,3 +111,12 @@ def __init__(self, name, level, message):
 
     def getMessage(self):
         return self.message
+
+
+class _Transport(object):
+
+    def __init__(self, client, name):
+        pass
+
+    def send(self, record, message):
+        self.send_called_with = (record, message)
diff --git a/gcloud/logging/handlers/transports/__init__.py b/gcloud/logging/handlers/transports/__init__.py
new file mode 100644
index 0000000000000..f9ca4239ddbe4
--- /dev/null
+++ b/gcloud/logging/handlers/transports/__init__.py
@@ -0,0 +1,26 @@
+# 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.
+
+"""Transport classes for Python logging integration.
+
+Currently two options are provided, a synchronous transport that makes
+an API call for each log statement, and an asynchronous handler that
+sends the API using a :class:`~gcloud.logging.logger.Batch` object in
+the background.
+"""
+
+from gcloud.logging.handlers.transports.base import Transport
+from gcloud.logging.handlers.transports.sync import SyncTransport
+from gcloud.logging.handlers.transports.background_thread import (
+    BackgroundThreadTransport)
diff --git a/gcloud/logging/handlers/transports/background_thread.py b/gcloud/logging/handlers/transports/background_thread.py
new file mode 100644
index 0000000000000..8909306cf55d5
--- /dev/null
+++ b/gcloud/logging/handlers/transports/background_thread.py
@@ -0,0 +1,171 @@
+# 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.
+
+"""Transport for Python logging handler
+
+Uses a background worker to log to Stackdriver Logging asynchronously.
+"""
+
+import atexit
+import copy
+import threading
+
+from gcloud.logging import Client
+from gcloud.logging.handlers.transports.base import Transport
+
+
+class _Worker(object):
+    """A threaded worker that writes batches of log entries
+
+    Writes entries to the logger API.
+
+    This class reuses a single :class:`Batch` method to write successive
+    entries.
+
+    Currently, the only public methods are constructing it (which also starts
+    it) and enqueuing :class:`Logger` (record, message) pairs.
+    """
+
+    def __init__(self, logger):
+        self.started = False
+        self.stopping = False
+        self.stopped = False
+
+        # _entries_condition is used to signal from the main thread whether
+        # there are any waiting queued logger entries to be written
+        self._entries_condition = threading.Condition()
+
+        # _stop_condition is used to signal from the worker thread to the
+        # main thread that it's finished its last entries
+        self._stop_condition = threading.Condition()
+
+        # This object continually reuses the same :class:`Batch` object to
+        # write multiple entries at the same time.
+        self.logger = logger
+        self.batch = self.logger.batch()
+
+        self._thread = None
+
+        # Number in seconds of  how long to wait for worker to send remaining
+        self._stop_timeout = 5
+
+        self._start()
+
+    def _run(self):
+        """The entry point for the worker thread.
+
+        Loops until ``stopping`` is set to :data:`True`, and commits batch
+        entries written during :meth:`enqueue`.
+        """
+        try:
+            self._entries_condition.acquire()
+            self.started = True
+            while not self.stopping:
+                if len(self.batch.entries) == 0:
+                    # branch coverage of this code extremely flaky
+                    self._entries_condition.wait()  # pragma: NO COVER
+
+                if len(self.batch.entries) > 0:
+                    self.batch.commit()
+        finally:
+            self._entries_condition.release()
+
+        # main thread may be waiting for worker thread to finish writing its
+        # final entries. here we signal that it's done.
+        self._stop_condition.acquire()
+        self._stop_condition.notify()
+        self._stop_condition.release()
+
+    def _start(self):
+        """Called by this class's constructor
+
+        This method is responsible for starting the thread and registering
+        the exit handlers.
+        """
+        try:
+            self._entries_condition.acquire()
+            self._thread = threading.Thread(
+                target=self._run,
+                name='gcloud.logging.handlers.transport.Worker')
+            self._thread.setDaemon(True)
+            self._thread.start()
+        finally:
+            self._entries_condition.release()
+            atexit.register(self._stop)
+
+    def _stop(self):
+        """Signals the worker thread to shut down
+
+        Also waits for ``stop_timeout`` seconds for the worker to finish.
+
+        This method is called by the ``atexit`` handler registered by
+         :meth:`start`.
+        """
+        if not self.started or self.stopping:
+            return
+
+        # lock the stop condition first so that the worker
+        # thread can't notify it's finished before we wait
+        self._stop_condition.acquire()
+
+        # now notify the worker thread to shutdown
+        self._entries_condition.acquire()
+        self.stopping = True
+        self._entries_condition.notify()
+        self._entries_condition.release()
+
+        # now wait for it to signal it's finished
+        self._stop_condition.wait(self._stop_timeout)
+        self._stop_condition.release()
+        self.stopped = True
+
+    def enqueue(self, record, message):
+        """Queues up a log entry to be written by the background thread."""
+        try:
+            self._entries_condition.acquire()
+            if self.stopping:
+                return
+            info = {'message': message, 'python_logger': record.name}
+            self.batch.log_struct(info, severity=record.levelname)
+            self._entries_condition.notify()
+        finally:
+            self._entries_condition.release()
+
+
+class BackgroundThreadTransport(Transport):
+    """Aysnchronous transport that uses a background thread.
+
+    Writes logging entries as a batch process.
+    """
+
+    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)
+        logger = self.client.logger(name)
+        self.worker = _Worker(logger)
+
+    def send(self, record, message):
+        """Overrides Transport.send().
+
+        :type record: :class:`logging.LogRecord`
+        :param record: Python log record that the handler was called with.
+
+        :type message: str
+        :param message: The message from the ``LogRecord`` after being
+                        formatted by the associated log formatters.
+        """
+        self.worker.enqueue(record, message)
diff --git a/gcloud/logging/handlers/transports/base.py b/gcloud/logging/handlers/transports/base.py
new file mode 100644
index 0000000000000..8f9d65647cb67
--- /dev/null
+++ b/gcloud/logging/handlers/transports/base.py
@@ -0,0 +1,35 @@
+# 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.
+
+"""Module containing base class for logging transport."""
+
+
+class Transport(object):
+    """Base class for ``gcloud`` logging handler transports.
+
+    Subclasses of :class:`Transport` must have constructors that accept a
+    client and name object, and must override :meth:`send`.
+    """
+
+    def send(self, record, message):
+        """Transport send to be implemented by subclasses.
+
+        :type record: :class:`logging.LogRecord`
+        :param record: Python log record that the handler was called with.
+
+        :type message: str
+        :param message: The message from the ``LogRecord`` after being
+                        formatted by the associated log formatters.
+        """
+        raise NotImplementedError
diff --git a/gcloud/logging/handlers/transports/sync.py b/gcloud/logging/handlers/transports/sync.py
new file mode 100644
index 0000000000000..afa39d311a194
--- /dev/null
+++ b/gcloud/logging/handlers/transports/sync.py
@@ -0,0 +1,43 @@
+# 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.
+
+"""Transport for Python logging handler.
+
+Logs directly to the the Stackdriver Logging API with a synchronous call.
+"""
+
+from gcloud.logging.handlers.transports.base import Transport
+
+
+class SyncTransport(Transport):
+    """Basic sychronous transport.
+
+    Uses this library's Logging client to directly make the API call.
+    """
+
+    def __init__(self, client, name):
+        self.logger = client.logger(name)
+
+    def send(self, record, message):
+        """Overrides transport.send().
+
+        :type record: :class:`logging.LogRecord`
+        :param record: Python log record that the handler was called with.
+
+        :type message: str
+        :param message: The message from the ``LogRecord`` after being
+                        formatted by the associated log formatters.
+        """
+        info = {'message': message, 'python_logger': record.name}
+        self.logger.log_struct(info, severity=record.levelname)
diff --git a/gcloud/logging/handlers/transports/test_background_thread.py b/gcloud/logging/handlers/transports/test_background_thread.py
new file mode 100644
index 0000000000000..5f9b5eb8ca32d
--- /dev/null
+++ b/gcloud/logging/handlers/transports/test_background_thread.py
@@ -0,0 +1,194 @@
+# 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 logging
+import time
+import unittest
+
+
+class TestBackgroundThreadHandler(unittest.TestCase):
+
+    PROJECT = 'PROJECT'
+
+    def _getTargetClass(self):
+        from gcloud.logging.handlers.transports import (
+            BackgroundThreadTransport)
+        return BackgroundThreadTransport
+
+    def _makeOne(self, *args, **kw):
+        return self._getTargetClass()(*args, **kw)
+
+    def test_ctor(self):
+        client = _Client(self.PROJECT)
+        NAME = 'python_logger'
+        transport = self._makeOne(client, NAME)
+        self.assertEquals(transport.worker.logger.name, NAME)
+
+    def test_send(self):
+        client = _Client(self.PROJECT)
+        NAME = 'python_logger'
+        transport = self._makeOne(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)
+
+        EXPECTED_STRUCT = {
+            'message': MESSAGE,
+            'python_logger': PYTHON_LOGGER_NAME
+        }
+        EXPECTED_SENT = (EXPECTED_STRUCT, logging.INFO)
+        self.assertEqual(transport.worker.batch.log_struct_called_with,
+                         EXPECTED_SENT)
+
+
+class TestWorker(unittest.TestCase):
+
+    def _getTargetClass(self):
+        from gcloud.logging.handlers.transports.background_thread import (
+            _Worker)
+        return _Worker
+
+    def _makeOne(self, *args, **kw):
+        return self._getTargetClass()(*args, **kw)
+
+    def test_ctor(self):
+        NAME = 'python_logger'
+        logger = _Logger(NAME)
+        worker = self._makeOne(logger)
+        self.assertEquals(worker.batch, logger._batch)
+
+    def test_run(self):
+        NAME = 'python_logger'
+        logger = _Logger(NAME)
+        worker = self._makeOne(logger)
+
+        PYTHON_LOGGER_NAME = 'mylogger'
+        MESSAGE = 'hello world'
+        record = _Record(PYTHON_LOGGER_NAME, logging.INFO, MESSAGE)
+
+        worker._start()
+
+        # first sleep is for branch coverage - ensure condition
+        # where queue is empty occurs
+        time.sleep(1)
+        # second polling is to avoid starting/stopping worker
+        # before anything ran
+        while not worker.started:
+            time.sleep(1)  # pragma: NO COVER
+
+        worker.enqueue(record, MESSAGE)
+        # Set timeout to none so worker thread finishes
+        worker._stop_timeout = None
+        worker._stop()
+        self.assertTrue(worker.batch.commit_called)
+
+    def test_run_after_stopped(self):
+        # No-op
+        NAME = 'python_logger'
+        logger = _Logger(NAME)
+        worker = self._makeOne(logger)
+
+        PYTHON_LOGGER_NAME = 'mylogger'
+        MESSAGE = 'hello world'
+        record = _Record(PYTHON_LOGGER_NAME, logging.INFO, MESSAGE)
+
+        worker._start()
+        while not worker.started:
+            time.sleep(1)  # pragma: NO COVER
+        worker._stop_timeout = None
+        worker._stop()
+        worker.enqueue(record, MESSAGE)
+        self.assertFalse(worker.batch.commit_called)
+        worker._stop()
+
+    def test_run_enqueue_early(self):
+        # No-op
+        NAME = 'python_logger'
+        logger = _Logger(NAME)
+        worker = self._makeOne(logger)
+
+        PYTHON_LOGGER_NAME = 'mylogger'
+        MESSAGE = 'hello world'
+        record = _Record(PYTHON_LOGGER_NAME, logging.INFO, MESSAGE)
+
+        worker.enqueue(record, MESSAGE)
+        worker._start()
+        while not worker.started:
+            time.sleep(1)  # pragma: NO COVER
+        worker._stop_timeout = None
+        worker._stop()
+        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):
+        self.entries = []
+        self.commit_called = False
+
+    def log_struct(self, record, severity=logging.INFO):
+        self.log_struct_called_with = (record, severity)
+        self.entries.append(record)
+
+    def commit(self):
+        self.commit_called = True
+        del self.entries[:]
+
+
+class _Credentials(object):
+
+    def authorize(self, _):
+        pass
+
+
+class _Connection(object):
+
+    def __init__(self):
+        self.http = None
+        self.credentials = _Credentials()
+
+
+class _Logger(object):
+
+    def __init__(self, name):
+        self.name = name
+
+    def batch(self):
+        self._batch = _Batch()
+        return self._batch
+
+
+class _Client(object):
+
+    def __init__(self, project):
+        self.project = project
+        self.connection = _Connection()
+
+    def logger(self, name):  # pylint: disable=unused-argument
+        self._logger = _Logger(name)
+        return self._logger
diff --git a/gcloud/logging/handlers/transports/test_base.py b/gcloud/logging/handlers/transports/test_base.py
new file mode 100644
index 0000000000000..daaaf4e1881de
--- /dev/null
+++ b/gcloud/logging/handlers/transports/test_base.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.
+
+import unittest
+
+
+class TestBaseHandler(unittest.TestCase):
+
+    PROJECT = 'PROJECT'
+
+    def _getTargetClass(self):
+        from gcloud.logging.handlers.transports import Transport
+        return Transport
+
+    def _makeOne(self, *args, **kw):
+        return self._getTargetClass()(*args, **kw)
+
+    def test_send_is_abstract(self):
+        client = _Client(self.PROJECT)
+        NAME = "python_logger"
+        target = self._makeOne(client, NAME)
+        self.assertRaises(NotImplementedError, lambda: target.send(None, None))
+
+
+class _Client(object):
+
+    def __init__(self, project):
+        self.project = project
diff --git a/gcloud/logging/handlers/transports/test_sync.py b/gcloud/logging/handlers/transports/test_sync.py
new file mode 100644
index 0000000000000..a252541948daf
--- /dev/null
+++ b/gcloud/logging/handlers/transports/test_sync.py
@@ -0,0 +1,93 @@
+# 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 logging
+import unittest
+
+
+class TestSyncHandler(unittest.TestCase):
+
+    PROJECT = 'PROJECT'
+
+    def _getTargetClass(self):
+        from gcloud.logging.handlers.transports import SyncTransport
+        return SyncTransport
+
+    def _makeOne(self, *args, **kw):
+        return self._getTargetClass()(*args, **kw)
+
+    def test_ctor(self):
+        client = _Client(self.PROJECT)
+        NAME = 'python_logger'
+        transport = self._makeOne(client, NAME)
+        self.assertEqual(transport.logger.name, 'python_logger')
+
+    def test_send(self):
+        client = _Client(self.PROJECT)
+        STACKDRIVER_LOGGER_NAME = 'python'
+        PYTHON_LOGGER_NAME = 'mylogger'
+        transport = self._makeOne(client, STACKDRIVER_LOGGER_NAME)
+        MESSAGE = 'hello world'
+        record = _Record(PYTHON_LOGGER_NAME, logging.INFO, MESSAGE)
+
+        transport.send(record, MESSAGE)
+        EXPECTED_STRUCT = {
+            'message': MESSAGE,
+            'python_logger': PYTHON_LOGGER_NAME
+        }
+        EXPECTED_SENT = (EXPECTED_STRUCT, logging.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):
+        self.name = name
+
+    def log_struct(self, message, severity=None):
+        self.log_struct_called_with = (message, severity)
+
+
+class _Client(object):
+
+    def __init__(self, project):
+        self.project = project
+
+    def logger(self, name):  # pylint: disable=unused-argument
+        self._logger = _Logger(name)
+        return self._logger
+
+
+class _Handler(object):
+
+    def __init__(self, level):
+        self.level = level  # pragma: NO COVER
+
+    def acquire(self):
+        pass  # pragma: NO COVER
+
+    def release(self):
+        pass  # pragma: NO COVER
diff --git a/gcloud/logging/logger.py b/gcloud/logging/logger.py
index 0c781c73c9d4d..066fe82a76f78 100644
--- a/gcloud/logging/logger.py
+++ b/gcloud/logging/logger.py
@@ -414,7 +414,7 @@ def commit(self, client=None):
             client = self.client
 
         kwargs = {
-            'logger_name': self.logger.path,
+            'logger_name': self.logger.full_name,
             'resource': {'type': 'global'},
         }
         if self.logger.labels is not None:
diff --git a/gcloud/logging/test_logger.py b/gcloud/logging/test_logger.py
index 4246b2137a267..31987266ba3e1 100644
--- a/gcloud/logging/test_logger.py
+++ b/gcloud/logging/test_logger.py
@@ -538,7 +538,7 @@ def test_commit_w_bound_client(self):
 
         self.assertEqual(list(batch.entries), [])
         self.assertEqual(api._write_entries_called_with,
-                         (ENTRIES, logger.path, RESOURCE, None))
+                         (ENTRIES, logger.full_name, RESOURCE, None))
 
     def test_commit_w_alternate_client(self):
         import json
@@ -582,7 +582,7 @@ def test_commit_w_alternate_client(self):
 
         self.assertEqual(list(batch.entries), [])
         self.assertEqual(api._write_entries_called_with,
-                         (ENTRIES, logger.path, RESOURCE, DEFAULT_LABELS))
+                         (ENTRIES, logger.full_name, RESOURCE, DEFAULT_LABELS))
 
     def test_context_mgr_success(self):
         import json
@@ -624,7 +624,7 @@ def test_context_mgr_success(self):
 
         self.assertEqual(list(batch.entries), [])
         self.assertEqual(api._write_entries_called_with,
-                         (ENTRIES, logger.path, RESOURCE, DEFAULT_LABELS))
+                         (ENTRIES, logger.full_name, RESOURCE, DEFAULT_LABELS))
 
     def test_context_mgr_failure(self):
         from google.protobuf.struct_pb2 import Struct, Value
@@ -670,7 +670,7 @@ class _Logger(object):
     labels = None
 
     def __init__(self, name="NAME", project="PROJECT"):
-        self.path = '/projects/%s/logs/%s' % (project, name)
+        self.full_name = 'projects/%s/logs/%s' % (project, name)
 
 
 class _DummyLoggingAPI(object):
diff --git a/scripts/verify_included_modules.py b/scripts/verify_included_modules.py
index 1172c0eb303d4..d351592ad9cf1 100644
--- a/scripts/verify_included_modules.py
+++ b/scripts/verify_included_modules.py
@@ -38,6 +38,8 @@
     'gcloud.error_reporting.__init__',
     'gcloud.iterator',
     'gcloud.logging.__init__',
+    'gcloud.logging.handlers.__init__',
+    'gcloud.logging.handlers.transports.__init__',
     'gcloud.monitoring.__init__',
     'gcloud.pubsub.__init__',
     'gcloud.resource_manager.__init__',
diff --git a/system_tests/logging_.py b/system_tests/logging_.py
index c3f57266d633d..c56bc729e3fec 100644
--- a/system_tests/logging_.py
+++ b/system_tests/logging_.py
@@ -15,16 +15,17 @@
 import logging
 import unittest
 
+import gcloud.logging
+import gcloud.logging.handlers.handlers
+from gcloud.logging.handlers.handlers import CloudLoggingHandler
+from gcloud.logging.handlers.transports import SyncTransport
 from gcloud import _helpers
 from gcloud.environment_vars import TESTS_PROJECT
-import gcloud.logging
-import gcloud.logging.handlers
 
 from retry import RetryErrors
 from retry import RetryResult
 from system_test_utils import unique_resource_id
 
-
 _RESOURCE_ID = unique_resource_id('-')
 DEFAULT_METRIC_NAME = 'system-tests-metric%s' % (_RESOURCE_ID,)
 DEFAULT_SINK_NAME = 'system-tests-sink%s' % (_RESOURCE_ID,)
@@ -136,34 +137,68 @@ def test_log_struct(self):
         self.assertEqual(len(entries), 1)
         self.assertEqual(entries[0].payload, JSON_PAYLOAD)
 
-    def test_log_handler(self):
+    def test_log_handler_async(self):
+        LOG_MESSAGE = 'It was the worst of times'
+
+        handler = CloudLoggingHandler(Config.CLIENT)
+        # only create the logger to delete, hidden otherwise
+        logger = Config.CLIENT.logger(handler.name)
+        self.to_delete.append(logger)
+
+        cloud_logger = logging.getLogger(handler.name)
+        cloud_logger.addHandler(handler)
+        cloud_logger.warn(LOG_MESSAGE)
+        entries, _ = self._list_entries(logger)
+        JSON_PAYLOAD = {
+            'message': LOG_MESSAGE,
+            'python_logger': handler.name
+        }
+        self.assertEqual(len(entries), 1)
+        self.assertEqual(entries[0].payload, JSON_PAYLOAD)
+
+    def test_log_handler_sync(self):
         LOG_MESSAGE = 'It was the best of times.'
+
+        handler = CloudLoggingHandler(Config.CLIENT,
+                                      name=self._logger_name(),
+                                      transport=SyncTransport)
+
         # only create the logger to delete, hidden otherwise
-        logger = Config.CLIENT.logger(self._logger_name())
+        logger = Config.CLIENT.logger(handler.name)
         self.to_delete.append(logger)
 
-        handler = gcloud.logging.handlers.CloudLoggingHandler(Config.CLIENT)
-        cloud_logger = logging.getLogger(self._logger_name())
+        LOGGER_NAME = 'mylogger'
+        cloud_logger = logging.getLogger(LOGGER_NAME)
         cloud_logger.addHandler(handler)
         cloud_logger.warn(LOG_MESSAGE)
 
         entries, _ = self._list_entries(logger)
+        JSON_PAYLOAD = {
+            'message': LOG_MESSAGE,
+            'python_logger': LOGGER_NAME
+        }
         self.assertEqual(len(entries), 1)
-        self.assertEqual(entries[0].payload, {'message': LOG_MESSAGE})
+        self.assertEqual(entries[0].payload, JSON_PAYLOAD)
 
     def test_log_root_handler(self):
         LOG_MESSAGE = 'It was the best of times.'
+
+        handler = CloudLoggingHandler(Config.CLIENT, name=self._logger_name())
         # only create the logger to delete, hidden otherwise
-        logger = Config.CLIENT.logger('root')
+        logger = Config.CLIENT.logger(handler.name)
         self.to_delete.append(logger)
 
-        handler = gcloud.logging.handlers.CloudLoggingHandler(Config.CLIENT)
-        gcloud.logging.handlers.setup_logging(handler)
+        gcloud.logging.handlers.handlers.setup_logging(handler)
         logging.warn(LOG_MESSAGE)
 
         entries, _ = self._list_entries(logger)
+        JSON_PAYLOAD = {
+            'message': LOG_MESSAGE,
+            'python_logger': 'root'
+        }
+
         self.assertEqual(len(entries), 1)
-        self.assertEqual(entries[0].payload, {'message': LOG_MESSAGE})
+        self.assertEqual(entries[0].payload, JSON_PAYLOAD)
 
     def test_log_struct_w_metadata(self):
         JSON_PAYLOAD = {