diff --git a/docs/index.rst b/docs/index.rst
index 64c5c65c7f..cabf56157c 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -23,6 +23,7 @@ API Documentation
 
   api-reference
   advanced-session-pool-topics
+  opentelemetry-tracing
 
 Changelog
 ---------
diff --git a/docs/opentelemetry-tracing.rst b/docs/opentelemetry-tracing.rst
new file mode 100644
index 0000000000..8906db43b6
--- /dev/null
+++ b/docs/opentelemetry-tracing.rst
@@ -0,0 +1,36 @@
+Tracing with OpenTelemetry
+==================================
+This library uses `OpenTelemetry <https://opentelemetry.io/>`_ to automatically generate traces providing insight on calls to Cloud Spanner. 
+For information on the benefits and utility of tracing, see the `Cloud Trace docs <https://cloud.google.com/trace/docs/overview>`_.
+
+To take advantage of these traces, we first need to install opentelemetry:
+
+.. code-block:: sh
+
+    pip install opentelemetry-api opentelemetry-sdk opentelemetry-instrumentation
+
+We also need to tell OpenTelemetry which exporter to use. For example, to export python-spanner traces to `Cloud Tracing <https://cloud.google.com/trace>`_, add the following lines to your application:
+
+.. code:: python
+
+    from opentelemetry import trace
+    from opentelemetry.sdk.trace import TracerProvider
+    from opentelemetry.trace.sampling import ProbabilitySampler
+    from opentelemetry.exporter.cloud_trace import CloudTraceSpanExporter
+    # BatchExportSpanProcessor exports spans to Cloud Trace 
+    # in a seperate thread to not block on the main thread
+    from opentelemetry.sdk.trace.export import BatchExportSpanProcessor
+
+    # Create and export one trace every 1000 requests
+    sampler = ProbabilitySampler(1/1000)
+    # Use the default tracer provider
+    trace.set_tracer_provider(TracerProvider(sampler=sampler))
+    trace.get_tracer_provider().add_span_processor(
+        # Initialize the cloud tracing exporter
+        BatchExportSpanProcessor(CloudTraceSpanExporter())
+    )
+
+Generated spanner traces should now be available on `Cloud Trace <https://console.cloud.google.com/traces>`_.
+
+Tracing is most effective when many libraries are instrumented to provide insight over the entire lifespan of a request.
+For a list of libraries that can be instrumented, see the `OpenTelemetry Integrations` section of the `OpenTelemetry Python docs <https://opentelemetry-python.readthedocs.io/en/stable/>`_
diff --git a/google/cloud/spanner_v1/_opentelemetry_tracing.py b/google/cloud/spanner_v1/_opentelemetry_tracing.py
new file mode 100644
index 0000000000..60e68598e9
--- /dev/null
+++ b/google/cloud/spanner_v1/_opentelemetry_tracing.py
@@ -0,0 +1,65 @@
+# Copyright 2020 Google LLC 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.
+
+"""Manages OpenTelemetry trace creation and handling"""
+
+from contextlib import contextmanager
+
+from google.api_core.exceptions import GoogleAPICallError
+from google.cloud.spanner_v1.gapic import spanner_client
+
+try:
+    from opentelemetry import trace
+    from opentelemetry.trace.status import Status, StatusCanonicalCode
+    from opentelemetry.instrumentation.utils import http_status_to_canonical_code
+
+    HAS_OPENTELEMETRY_INSTALLED = True
+except ImportError:
+    HAS_OPENTELEMETRY_INSTALLED = False
+
+
+@contextmanager
+def trace_call(name, session, extra_attributes=None):
+    if not HAS_OPENTELEMETRY_INSTALLED or not session:
+        # Empty context manager. Users will have to check if the generated value is None or a span
+        yield None
+        return
+
+    tracer = trace.get_tracer(__name__)
+
+    # Set base attributes that we know for every trace created
+    attributes = {
+        "db.type": "spanner",
+        "db.url": spanner_client.SpannerClient.SERVICE_ADDRESS,
+        "db.instance": session._database.name,
+        "net.host.name": spanner_client.SpannerClient.SERVICE_ADDRESS,
+    }
+
+    if extra_attributes:
+        attributes.update(extra_attributes)
+
+    with tracer.start_as_current_span(
+        name, kind=trace.SpanKind.CLIENT, attributes=attributes
+    ) as span:
+        try:
+            yield span
+        except GoogleAPICallError as error:
+            if error.code is not None:
+                span.set_status(Status(http_status_to_canonical_code(error.code)))
+            elif error.grpc_status_code is not None:
+                span.set_status(
+                    # OpenTelemetry's StatusCanonicalCode maps 1-1 with grpc status codes
+                    Status(StatusCanonicalCode(error.grpc_status_code.value[0]))
+                )
+            raise
diff --git a/google/cloud/spanner_v1/batch.py b/google/cloud/spanner_v1/batch.py
index e62763d7fd..7ab394b215 100644
--- a/google/cloud/spanner_v1/batch.py
+++ b/google/cloud/spanner_v1/batch.py
@@ -22,6 +22,7 @@
 from google.cloud.spanner_v1._helpers import _SessionWrapper
 from google.cloud.spanner_v1._helpers import _make_list_value_pbs
 from google.cloud.spanner_v1._helpers import _metadata_with_prefix
+from google.cloud.spanner_v1._opentelemetry_tracing import trace_call
 
 # pylint: enable=ungrouped-imports
 
@@ -147,12 +148,14 @@ def commit(self):
         api = database.spanner_api
         metadata = _metadata_with_prefix(database.name)
         txn_options = TransactionOptions(read_write=TransactionOptions.ReadWrite())
-        response = api.commit(
-            self._session.name,
-            mutations=self._mutations,
-            single_use_transaction=txn_options,
-            metadata=metadata,
-        )
+        trace_attributes = {"num_mutations": len(self._mutations)}
+        with trace_call("CloudSpanner.Commit", self._session, trace_attributes):
+            response = api.commit(
+                self._session.name,
+                mutations=self._mutations,
+                single_use_transaction=txn_options,
+                metadata=metadata,
+            )
         self.committed = _pb_timestamp_to_datetime(response.commit_timestamp)
         return self.committed
 
diff --git a/google/cloud/spanner_v1/session.py b/google/cloud/spanner_v1/session.py
index a84aaa7c6d..b3a1b7e6d8 100644
--- a/google/cloud/spanner_v1/session.py
+++ b/google/cloud/spanner_v1/session.py
@@ -26,6 +26,7 @@
 from google.cloud.spanner_v1.batch import Batch
 from google.cloud.spanner_v1.snapshot import Snapshot
 from google.cloud.spanner_v1.transaction import Transaction
+from google.cloud.spanner_v1._opentelemetry_tracing import trace_call
 import random
 
 # pylint: enable=ungrouped-imports
@@ -114,7 +115,11 @@ def create(self):
         kw = {}
         if self._labels:
             kw = {"session": {"labels": self._labels}}
-        session_pb = api.create_session(self._database.name, metadata=metadata, **kw)
+
+        with trace_call("CloudSpanner.CreateSession", self, self._labels):
+            session_pb = api.create_session(
+                self._database.name, metadata=metadata, **kw
+            )
         self._session_id = session_pb.name.split("/")[-1]
 
     def exists(self):
@@ -130,10 +135,16 @@ def exists(self):
             return False
         api = self._database.spanner_api
         metadata = _metadata_with_prefix(self._database.name)
-        try:
-            api.get_session(self.name, metadata=metadata)
-        except NotFound:
-            return False
+
+        with trace_call("CloudSpanner.GetSession", self) as span:
+            try:
+                api.get_session(self.name, metadata=metadata)
+                if span:
+                    span.set_attribute("session_found", True)
+            except NotFound:
+                if span:
+                    span.set_attribute("session_found", False)
+                return False
 
         return True
 
@@ -150,8 +161,8 @@ def delete(self):
             raise ValueError("Session ID not set by back-end")
         api = self._database.spanner_api
         metadata = _metadata_with_prefix(self._database.name)
-
-        api.delete_session(self.name, metadata=metadata)
+        with trace_call("CloudSpanner.DeleteSession", self):
+            api.delete_session(self.name, metadata=metadata)
 
     def ping(self):
         """Ping the session to keep it alive by executing "SELECT 1".
diff --git a/google/cloud/spanner_v1/snapshot.py b/google/cloud/spanner_v1/snapshot.py
index dcb6e32d88..0b5ee1d894 100644
--- a/google/cloud/spanner_v1/snapshot.py
+++ b/google/cloud/spanner_v1/snapshot.py
@@ -30,9 +30,10 @@
 from google.cloud.spanner_v1._helpers import _SessionWrapper
 from google.cloud.spanner_v1.streamed import StreamedResultSet
 from google.cloud.spanner_v1.types import PartitionOptions
+from google.cloud.spanner_v1._opentelemetry_tracing import trace_call
 
 
-def _restart_on_unavailable(restart):
+def _restart_on_unavailable(restart, trace_name=None, session=None, attributes=None):
     """Restart iteration after :exc:`.ServiceUnavailable`.
 
     :type restart: callable
@@ -40,7 +41,8 @@ def _restart_on_unavailable(restart):
     """
     resume_token = b""
     item_buffer = []
-    iterator = restart()
+    with trace_call(trace_name, session, attributes):
+        iterator = restart()
     while True:
         try:
             for item in iterator:
@@ -50,7 +52,8 @@ def _restart_on_unavailable(restart):
                     break
         except ServiceUnavailable:
             del item_buffer[:]
-            iterator = restart(resume_token=resume_token)
+            with trace_call(trace_name, session, attributes):
+                iterator = restart(resume_token=resume_token)
             continue
 
         if len(item_buffer) == 0:
@@ -143,7 +146,10 @@ def read(self, table, columns, keyset, index="", limit=0, partition=None):
             metadata=metadata,
         )
 
-        iterator = _restart_on_unavailable(restart)
+        trace_attributes = {"table_id": table, "columns": columns}
+        iterator = _restart_on_unavailable(
+            restart, "CloudSpanner.ReadOnlyTransaction", self._session, trace_attributes
+        )
 
         self._read_request_count += 1
 
@@ -243,7 +249,13 @@ def execute_sql(
             timeout=timeout,
         )
 
-        iterator = _restart_on_unavailable(restart)
+        trace_attributes = {"db.statement": sql}
+        iterator = _restart_on_unavailable(
+            restart,
+            "CloudSpanner.ReadWriteTransaction",
+            self._session,
+            trace_attributes,
+        )
 
         self._read_request_count += 1
         self._execute_sql_count += 1
@@ -309,16 +321,20 @@ def partition_read(
             partition_size_bytes=partition_size_bytes, max_partitions=max_partitions
         )
 
-        response = api.partition_read(
-            session=self._session.name,
-            table=table,
-            columns=columns,
-            key_set=keyset._to_pb(),
-            transaction=transaction,
-            index=index,
-            partition_options=partition_options,
-            metadata=metadata,
-        )
+        trace_attributes = {"table_id": table, "columns": columns}
+        with trace_call(
+            "CloudSpanner.PartitionReadOnlyTransaction", self._session, trace_attributes
+        ):
+            response = api.partition_read(
+                session=self._session.name,
+                table=table,
+                columns=columns,
+                key_set=keyset._to_pb(),
+                transaction=transaction,
+                index=index,
+                partition_options=partition_options,
+                metadata=metadata,
+            )
 
         return [partition.partition_token for partition in response.partitions]
 
@@ -385,15 +401,21 @@ def partition_query(
             partition_size_bytes=partition_size_bytes, max_partitions=max_partitions
         )
 
-        response = api.partition_query(
-            session=self._session.name,
-            sql=sql,
-            transaction=transaction,
-            params=params_pb,
-            param_types=param_types,
-            partition_options=partition_options,
-            metadata=metadata,
-        )
+        trace_attributes = {"db.statement": sql}
+        with trace_call(
+            "CloudSpanner.PartitionReadWriteTransaction",
+            self._session,
+            trace_attributes,
+        ):
+            response = api.partition_query(
+                session=self._session.name,
+                sql=sql,
+                transaction=transaction,
+                params=params_pb,
+                param_types=param_types,
+                partition_options=partition_options,
+                metadata=metadata,
+            )
 
         return [partition.partition_token for partition in response.partitions]
 
@@ -515,8 +537,9 @@ def begin(self):
         api = database.spanner_api
         metadata = _metadata_with_prefix(database.name)
         txn_selector = self._make_txn_selector()
-        response = api.begin_transaction(
-            self._session.name, txn_selector.begin, metadata=metadata
-        )
+        with trace_call("CloudSpanner.BeginTransaction", self._session):
+            response = api.begin_transaction(
+                self._session.name, txn_selector.begin, metadata=metadata
+            )
         self._transaction_id = response.id
         return self._transaction_id
diff --git a/google/cloud/spanner_v1/transaction.py b/google/cloud/spanner_v1/transaction.py
index 3c1abc7326..40116a9bbb 100644
--- a/google/cloud/spanner_v1/transaction.py
+++ b/google/cloud/spanner_v1/transaction.py
@@ -26,6 +26,7 @@
 from google.cloud.spanner_v1.proto.transaction_pb2 import TransactionOptions
 from google.cloud.spanner_v1.snapshot import _SnapshotBase
 from google.cloud.spanner_v1.batch import _BatchBase
+from google.cloud.spanner_v1._opentelemetry_tracing import trace_call
 
 
 class Transaction(_SnapshotBase, _BatchBase):
@@ -95,9 +96,10 @@ def begin(self):
         api = database.spanner_api
         metadata = _metadata_with_prefix(database.name)
         txn_options = TransactionOptions(read_write=TransactionOptions.ReadWrite())
-        response = api.begin_transaction(
-            self._session.name, txn_options, metadata=metadata
-        )
+        with trace_call("CloudSpanner.BeginTransaction", self._session):
+            response = api.begin_transaction(
+                self._session.name, txn_options, metadata=metadata
+            )
         self._transaction_id = response.id
         return self._transaction_id
 
@@ -107,7 +109,8 @@ def rollback(self):
         database = self._session._database
         api = database.spanner_api
         metadata = _metadata_with_prefix(database.name)
-        api.rollback(self._session.name, self._transaction_id, metadata=metadata)
+        with trace_call("CloudSpanner.Rollback", self._session):
+            api.rollback(self._session.name, self._transaction_id, metadata=metadata)
         self.rolled_back = True
         del self._session._transaction
 
@@ -123,12 +126,14 @@ def commit(self):
         database = self._session._database
         api = database.spanner_api
         metadata = _metadata_with_prefix(database.name)
-        response = api.commit(
-            self._session.name,
-            mutations=self._mutations,
-            transaction_id=self._transaction_id,
-            metadata=metadata,
-        )
+        trace_attributes = {"num_mutations": len(self._mutations)}
+        with trace_call("CloudSpanner.Commit", self._session, trace_attributes):
+            response = api.commit(
+                self._session.name,
+                mutations=self._mutations,
+                transaction_id=self._transaction_id,
+                metadata=metadata,
+            )
         self.committed = _pb_timestamp_to_datetime(response.commit_timestamp)
         del self._session._transaction
         return self.committed
@@ -212,17 +217,21 @@ def execute_update(
         default_query_options = database._instance._client._query_options
         query_options = _merge_query_options(default_query_options, query_options)
 
-        response = api.execute_sql(
-            self._session.name,
-            dml,
-            transaction=transaction,
-            params=params_pb,
-            param_types=param_types,
-            query_mode=query_mode,
-            query_options=query_options,
-            seqno=seqno,
-            metadata=metadata,
-        )
+        trace_attributes = {"db.statement": dml}
+        with trace_call(
+            "CloudSpanner.ReadWriteTransaction", self._session, trace_attributes
+        ):
+            response = api.execute_sql(
+                self._session.name,
+                dml,
+                transaction=transaction,
+                params=params_pb,
+                param_types=param_types,
+                query_mode=query_mode,
+                query_options=query_options,
+                seqno=seqno,
+                metadata=metadata,
+            )
         return response.stats.row_count_exact
 
     def batch_update(self, statements):
@@ -268,13 +277,18 @@ def batch_update(self, statements):
             self._execute_sql_count + 1,
         )
 
-        response = api.execute_batch_dml(
-            session=self._session.name,
-            transaction=transaction,
-            statements=parsed,
-            seqno=seqno,
-            metadata=metadata,
-        )
+        trace_attributes = {
+            # Get just the queries from the DML statement batch
+            "db.statement": ";".join([statement["sql"] for statement in parsed])
+        }
+        with trace_call("CloudSpanner.DMLTransaction", self._session, trace_attributes):
+            response = api.execute_batch_dml(
+                session=self._session.name,
+                transaction=transaction,
+                statements=parsed,
+                seqno=seqno,
+                metadata=metadata,
+            )
         row_counts = [
             result_set.stats.row_count_exact for result_set in response.result_sets
         ]
diff --git a/noxfile.py b/noxfile.py
index ee0e4c8b78..91de61a9de 100644
--- a/noxfile.py
+++ b/noxfile.py
@@ -66,6 +66,13 @@ def lint_setup_py(session):
 def default(session):
     # Install all test dependencies, then install this package in-place.
     session.install("mock", "pytest", "pytest-cov")
+
+    # Install opentelemetry dependencies if python3+
+    if session.python != "2.7":
+        session.install(
+            "opentelemetry-api", "opentelemetry-sdk", "opentelemetry-instrumentation"
+        )
+
     session.install("-e", ".")
 
     # Run py.test against the unit tests.
@@ -115,6 +122,12 @@ def system(session):
     # virtualenv's dist-packages.
     session.install("mock", "pytest")
 
+    # Install opentelemetry dependencies if not 2.7
+    if session.python != "2.7":
+        session.install(
+            "opentelemetry-api", "opentelemetry-sdk", "opentelemetry-instrumentation"
+        )
+
     session.install("-e", ".")
     session.install("-e", "test_utils/")
 
diff --git a/tests/_helpers.py b/tests/_helpers.py
new file mode 100644
index 0000000000..6ebc4bb374
--- /dev/null
+++ b/tests/_helpers.py
@@ -0,0 +1,50 @@
+import unittest
+import mock
+
+try:
+    from opentelemetry import trace as trace_api
+    from opentelemetry.trace.status import StatusCanonicalCode
+
+    from opentelemetry.sdk.trace import TracerProvider, export
+    from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
+        InMemorySpanExporter,
+    )
+
+    HAS_OPENTELEMETRY_INSTALLED = True
+except ImportError:
+    HAS_OPENTELEMETRY_INSTALLED = False
+
+    StatusCanonicalCode = mock.Mock()
+
+
+class OpenTelemetryBase(unittest.TestCase):
+    def setUp(self):
+        if HAS_OPENTELEMETRY_INSTALLED:
+            self.original_tracer_provider = trace_api.get_tracer_provider()
+            self.tracer_provider = TracerProvider()
+            self.memory_exporter = InMemorySpanExporter()
+            span_processor = export.SimpleExportSpanProcessor(self.memory_exporter)
+            self.tracer_provider.add_span_processor(span_processor)
+            trace_api.set_tracer_provider(self.tracer_provider)
+
+    def tearDown(self):
+        if HAS_OPENTELEMETRY_INSTALLED:
+            trace_api.set_tracer_provider(self.original_tracer_provider)
+
+    def assertNoSpans(self):
+        if HAS_OPENTELEMETRY_INSTALLED:
+            span_list = self.memory_exporter.get_finished_spans()
+            self.assertEqual(len(span_list), 0)
+
+    def assertSpanAttributes(
+        self, name, status=StatusCanonicalCode.OK, attributes=None, span=None
+    ):
+        if HAS_OPENTELEMETRY_INSTALLED:
+            if not span:
+                span_list = self.memory_exporter.get_finished_spans()
+                self.assertEqual(len(span_list), 1)
+                span = span_list[0]
+
+            self.assertEqual(span.name, name)
+            self.assertEqual(span.status.canonical_code, status)
+            self.assertEqual(span.attributes, attributes)
diff --git a/tests/system/test_system.py b/tests/system/test_system.py
index 9fde7db0c3..7779769c8f 100644
--- a/tests/system/test_system.py
+++ b/tests/system/test_system.py
@@ -52,6 +52,7 @@
 from test_utils.retry import RetryResult
 from test_utils.system import unique_resource_id
 from tests._fixtures import DDL_STATEMENTS
+from tests._helpers import OpenTelemetryBase, HAS_OPENTELEMETRY_INSTALLED
 
 
 CREATE_INSTANCE = os.getenv("GOOGLE_CLOUD_TESTS_CREATE_SPANNER_INSTANCE") is not None
@@ -67,6 +68,12 @@
 COUNTERS_TABLE = "counters"
 COUNTERS_COLUMNS = ("name", "value")
 
+BASE_ATTRIBUTES = {
+    "db.type": "spanner",
+    "db.url": "spanner.googleapis.com:443",
+    "net.host.name": "spanner.googleapis.com:443",
+}
+
 _STATUS_CODE_TO_GRPC_STATUS_CODE = {
     member.value[0]: member for member in grpc.StatusCode
 }
@@ -726,7 +733,7 @@ def test_list_backups(self):
 NANO_TIME = DatetimeWithNanoseconds(1995, 8, 31, nanosecond=987654321)
 POS_INF = float("+inf")
 NEG_INF = float("-inf")
-OTHER_NAN, = struct.unpack("<d", b"\x01\x00\x01\x00\x00\x00\xf8\xff")
+(OTHER_NAN,) = struct.unpack("<d", b"\x01\x00\x01\x00\x00\x00\xf8\xff")
 BYTES_1 = b"Ymlu"
 BYTES_2 = b"Ym9vdHM="
 ALL_TYPES_TABLE = "all_types"
@@ -781,7 +788,7 @@ def test_list_backups(self):
 )
 
 
-class TestSessionAPI(unittest.TestCase, _TestData):
+class TestSessionAPI(OpenTelemetryBase, _TestData):
     DATABASE_NAME = "test_sessions" + unique_resource_id("_")
 
     @classmethod
@@ -798,9 +805,11 @@ def tearDownClass(cls):
         cls._db.drop()
 
     def setUp(self):
+        super(TestSessionAPI, self).setUp()
         self.to_delete = []
 
     def tearDown(self):
+        super(TestSessionAPI, self).tearDown()
         for doomed in self.to_delete:
             doomed.delete()
 
@@ -826,6 +835,46 @@ def test_batch_insert_then_read(self):
             rows = list(snapshot.read(self.TABLE, self.COLUMNS, self.ALL))
         self._check_rows_data(rows)
 
+        if HAS_OPENTELEMETRY_INSTALLED:
+            span_list = self.memory_exporter.get_finished_spans()
+            self.assertEqual(len(span_list), 4)
+            self.assertSpanAttributes(
+                "CloudSpanner.GetSession",
+                attributes=dict(
+                    BASE_ATTRIBUTES,
+                    **{"db.instance": self._db.name, "session_found": True}
+                ),
+                span=span_list[0],
+            )
+            self.assertSpanAttributes(
+                "CloudSpanner.Commit",
+                attributes=dict(
+                    BASE_ATTRIBUTES,
+                    **{"db.instance": self._db.name, "num_mutations": 2}
+                ),
+                span=span_list[1],
+            )
+            self.assertSpanAttributes(
+                "CloudSpanner.GetSession",
+                attributes=dict(
+                    BASE_ATTRIBUTES,
+                    **{"db.instance": self._db.name, "session_found": True}
+                ),
+                span=span_list[2],
+            )
+            self.assertSpanAttributes(
+                "CloudSpanner.ReadOnlyTransaction",
+                attributes=dict(
+                    BASE_ATTRIBUTES,
+                    **{
+                        "db.instance": self._db.name,
+                        "columns": self.COLUMNS,
+                        "table_id": self.TABLE,
+                    }
+                ),
+                span=span_list[3],
+            )
+
     def test_batch_insert_then_read_string_array_of_string(self):
         TABLE = "string_plus_array_of_string"
         COLUMNS = ["id", "name", "tags"]
@@ -924,6 +973,77 @@ def test_transaction_read_and_insert_then_rollback(self):
         rows = list(session.read(self.TABLE, self.COLUMNS, self.ALL))
         self.assertEqual(rows, [])
 
+        if HAS_OPENTELEMETRY_INSTALLED:
+            span_list = self.memory_exporter.get_finished_spans()
+            self.assertEqual(len(span_list), 8)
+            self.assertSpanAttributes(
+                "CloudSpanner.CreateSession",
+                attributes=dict(BASE_ATTRIBUTES, **{"db.instance": self._db.name}),
+                span=span_list[0],
+            )
+            self.assertSpanAttributes(
+                "CloudSpanner.GetSession",
+                attributes=dict(
+                    BASE_ATTRIBUTES,
+                    **{"db.instance": self._db.name, "session_found": True}
+                ),
+                span=span_list[1],
+            )
+            self.assertSpanAttributes(
+                "CloudSpanner.Commit",
+                attributes=dict(
+                    BASE_ATTRIBUTES,
+                    **{"db.instance": self._db.name, "num_mutations": 1}
+                ),
+                span=span_list[2],
+            )
+            self.assertSpanAttributes(
+                "CloudSpanner.BeginTransaction",
+                attributes=dict(BASE_ATTRIBUTES, **{"db.instance": self._db.name}),
+                span=span_list[3],
+            )
+            self.assertSpanAttributes(
+                "CloudSpanner.ReadOnlyTransaction",
+                attributes=dict(
+                    BASE_ATTRIBUTES,
+                    **{
+                        "db.instance": self._db.name,
+                        "table_id": self.TABLE,
+                        "columns": self.COLUMNS,
+                    }
+                ),
+                span=span_list[4],
+            )
+            self.assertSpanAttributes(
+                "CloudSpanner.ReadOnlyTransaction",
+                attributes=dict(
+                    BASE_ATTRIBUTES,
+                    **{
+                        "db.instance": self._db.name,
+                        "table_id": self.TABLE,
+                        "columns": self.COLUMNS,
+                    }
+                ),
+                span=span_list[5],
+            )
+            self.assertSpanAttributes(
+                "CloudSpanner.Rollback",
+                attributes=dict(BASE_ATTRIBUTES, **{"db.instance": self._db.name}),
+                span=span_list[6],
+            )
+            self.assertSpanAttributes(
+                "CloudSpanner.ReadOnlyTransaction",
+                attributes=dict(
+                    BASE_ATTRIBUTES,
+                    **{
+                        "db.instance": self._db.name,
+                        "table_id": self.TABLE,
+                        "columns": self.COLUMNS,
+                    }
+                ),
+                span=span_list[7],
+            )
+
     def _transaction_read_then_raise(self, transaction):
         rows = list(transaction.read(self.TABLE, self.COLUMNS, self.ALL))
         self.assertEqual(len(rows), 0)
@@ -1226,6 +1346,66 @@ def test_transaction_batch_update_wo_statements(self):
             with self.assertRaises(InvalidArgument):
                 transaction.batch_update([])
 
+    def test_transaction_batch_update_w_parent_span(self):
+        try:
+            from opentelemetry import trace
+        except ImportError:
+            return
+
+        tracer = trace.get_tracer(__name__)
+
+        retry = RetryInstanceState(_has_all_ddl)
+        retry(self._db.reload)()
+
+        session = self._db.session()
+        session.create()
+        self.to_delete.append(session)
+
+        with session.batch() as batch:
+            batch.delete(self.TABLE, self.ALL)
+
+        insert_statement = list(self._generate_insert_statements())[0]
+        update_statement = (
+            "UPDATE contacts SET email = @email " "WHERE contact_id = @contact_id;",
+            {"contact_id": 1, "email": "phreddy@example.com"},
+            {"contact_id": Type(code=INT64), "email": Type(code=STRING)},
+        )
+        delete_statement = (
+            "DELETE contacts WHERE contact_id = @contact_id;",
+            {"contact_id": 1},
+            {"contact_id": Type(code=INT64)},
+        )
+
+        def unit_of_work(transaction, self):
+
+            status, row_counts = transaction.batch_update(
+                [insert_statement, update_statement, delete_statement]
+            )
+            self._check_batch_status(status.code)
+            self.assertEqual(len(row_counts), 3)
+            for row_count in row_counts:
+                self.assertEqual(row_count, 1)
+
+        with tracer.start_as_current_span("Test Span"):
+            session.run_in_transaction(unit_of_work, self)
+
+        span_list = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(span_list), 6)
+        self.assertEqual(
+            list(map(lambda span: span.name, span_list)),
+            [
+                "CloudSpanner.CreateSession",
+                "CloudSpanner.Commit",
+                "CloudSpanner.BeginTransaction",
+                "CloudSpanner.DMLTransaction",
+                "CloudSpanner.Commit",
+                "Test Span",
+            ],
+        )
+        for span in span_list[2:-1]:
+            self.assertEqual(span.context.trace_id, span_list[-1].context.trace_id)
+            self.assertEqual(span.parent.span_id, span_list[-1].context.span_id)
+
     def test_execute_partitioned_dml(self):
         # [START spanner_test_dml_partioned_dml_update]
         retry = RetryInstanceState(_has_all_ddl)
@@ -2333,7 +2513,7 @@ def test_execute_sql_returning_transfinite_floats(self):
                 )
             )
             self.assertEqual(len(rows), 1)
-            float_array, = rows[0]
+            float_array = rows[0][0]
             self.assertEqual(float_array[0], float("-inf"))
             self.assertEqual(float_array[1], float("+inf"))
             # NaNs cannot be searched for by equality.
diff --git a/tests/unit/test__opentelemetry_tracing.py b/tests/unit/test__opentelemetry_tracing.py
new file mode 100644
index 0000000000..8e26468dfe
--- /dev/null
+++ b/tests/unit/test__opentelemetry_tracing.py
@@ -0,0 +1,129 @@
+import importlib
+import mock
+import unittest
+import sys
+
+try:
+    from opentelemetry import trace as trace_api
+    from opentelemetry.trace.status import StatusCanonicalCode
+except ImportError:
+    pass
+
+from google.api_core.exceptions import GoogleAPICallError
+from google.cloud.spanner_v1 import _opentelemetry_tracing
+
+from tests._helpers import OpenTelemetryBase, HAS_OPENTELEMETRY_INSTALLED
+
+
+def _make_rpc_error(error_cls, trailing_metadata=None):
+    import grpc
+
+    grpc_error = mock.create_autospec(grpc.Call, instance=True)
+    grpc_error.trailing_metadata.return_value = trailing_metadata
+    return error_cls("error", errors=(grpc_error,))
+
+
+def _make_session():
+    from google.cloud.spanner_v1.session import Session
+
+    return mock.Mock(autospec=Session, instance=True)
+
+
+# Skip all of these tests if we don't have OpenTelemetry
+if HAS_OPENTELEMETRY_INSTALLED:
+
+    class TestNoTracing(unittest.TestCase):
+        def setUp(self):
+            self._temp_opentelemetry = sys.modules["opentelemetry"]
+
+            sys.modules["opentelemetry"] = None
+            importlib.reload(_opentelemetry_tracing)
+
+        def tearDown(self):
+            sys.modules["opentelemetry"] = self._temp_opentelemetry
+            importlib.reload(_opentelemetry_tracing)
+
+        def test_no_trace_call(self):
+            with _opentelemetry_tracing.trace_call("Test", _make_session()) as no_span:
+                self.assertIsNone(no_span)
+
+    class TestTracing(OpenTelemetryBase):
+        def test_trace_call(self):
+            extra_attributes = {
+                "attribute1": "value1",
+                # Since our database is mocked, we have to override the db.instance parameter so it is a string
+                "db.instance": "database_name",
+            }
+
+            expected_attributes = {
+                "db.type": "spanner",
+                "db.url": "spanner.googleapis.com:443",
+                "net.host.name": "spanner.googleapis.com:443",
+            }
+            expected_attributes.update(extra_attributes)
+
+            with _opentelemetry_tracing.trace_call(
+                "CloudSpanner.Test", _make_session(), extra_attributes
+            ) as span:
+                span.set_attribute("after_setup_attribute", 1)
+
+            expected_attributes["after_setup_attribute"] = 1
+
+            span_list = self.memory_exporter.get_finished_spans()
+            self.assertEqual(len(span_list), 1)
+            span = span_list[0]
+            self.assertEqual(span.kind, trace_api.SpanKind.CLIENT)
+            self.assertEqual(span.attributes, expected_attributes)
+            self.assertEqual(span.name, "CloudSpanner.Test")
+            self.assertEqual(span.status.canonical_code, StatusCanonicalCode.OK)
+
+        def test_trace_error(self):
+            extra_attributes = {"db.instance": "database_name"}
+
+            expected_attributes = {
+                "db.type": "spanner",
+                "db.url": "spanner.googleapis.com:443",
+                "net.host.name": "spanner.googleapis.com:443",
+            }
+            expected_attributes.update(extra_attributes)
+
+            with self.assertRaises(GoogleAPICallError):
+                with _opentelemetry_tracing.trace_call(
+                    "CloudSpanner.Test", _make_session(), extra_attributes
+                ) as span:
+                    from google.api_core.exceptions import InvalidArgument
+
+                    raise _make_rpc_error(InvalidArgument)
+
+            span_list = self.memory_exporter.get_finished_spans()
+            self.assertEqual(len(span_list), 1)
+            span = span_list[0]
+            self.assertEqual(span.kind, trace_api.SpanKind.CLIENT)
+            self.assertEqual(span.attributes, expected_attributes)
+            self.assertEqual(span.name, "CloudSpanner.Test")
+            self.assertEqual(
+                span.status.canonical_code, StatusCanonicalCode.INVALID_ARGUMENT
+            )
+
+        def test_trace_grpc_error(self):
+            extra_attributes = {"db.instance": "database_name"}
+
+            expected_attributes = {
+                "db.type": "spanner",
+                "db.url": "spanner.googleapis.com:443",
+                "net.host.name": "spanner.googleapis.com:443",
+            }
+            expected_attributes.update(extra_attributes)
+
+            with self.assertRaises(GoogleAPICallError):
+                with _opentelemetry_tracing.trace_call(
+                    "CloudSpanner.Test", _make_session(), extra_attributes
+                ) as span:
+                    from google.api_core.exceptions import DataLoss
+
+                    raise _make_rpc_error(DataLoss)
+
+            span_list = self.memory_exporter.get_finished_spans()
+            self.assertEqual(len(span_list), 1)
+            span = span_list[0]
+            self.assertEqual(span.status.canonical_code, StatusCanonicalCode.DATA_LOSS)
diff --git a/tests/unit/test_batch.py b/tests/unit/test_batch.py
index f4fd9d12ab..9b831f4906 100644
--- a/tests/unit/test_batch.py
+++ b/tests/unit/test_batch.py
@@ -14,7 +14,7 @@
 
 
 import unittest
-
+from tests._helpers import OpenTelemetryBase, StatusCanonicalCode
 
 TABLE_NAME = "citizens"
 COLUMNS = ["email", "first_name", "last_name", "age"]
@@ -22,6 +22,12 @@
     [u"phred@exammple.com", u"Phred", u"Phlyntstone", 32],
     [u"bharney@example.com", u"Bharney", u"Rhubble", 31],
 ]
+BASE_ATTRIBUTES = {
+    "db.type": "spanner",
+    "db.url": "spanner.googleapis.com:443",
+    "db.instance": "testing",
+    "net.host.name": "spanner.googleapis.com:443",
+}
 
 
 class _BaseTest(unittest.TestCase):
@@ -164,7 +170,7 @@ def test_delete(self):
             )
 
 
-class TestBatch(_BaseTest):
+class TestBatch(_BaseTest, OpenTelemetryBase):
     def _getTargetClass(self):
         from google.cloud.spanner_v1.batch import Batch
 
@@ -189,6 +195,8 @@ def test_commit_already_committed(self):
         with self.assertRaises(ValueError):
             batch.commit()
 
+        self.assertNoSpans()
+
     def test_commit_grpc_error(self):
         from google.api_core.exceptions import Unknown
         from google.cloud.spanner_v1.keyset import KeySet
@@ -204,6 +212,12 @@ def test_commit_grpc_error(self):
         with self.assertRaises(Unknown):
             batch.commit()
 
+        self.assertSpanAttributes(
+            "CloudSpanner.Commit",
+            status=StatusCanonicalCode.UNKNOWN,
+            attributes=dict(BASE_ATTRIBUTES, num_mutations=1),
+        )
+
     def test_commit_ok(self):
         import datetime
         from google.cloud.spanner_v1.proto.spanner_pb2 import CommitResponse
@@ -232,6 +246,10 @@ def test_commit_ok(self):
         self.assertTrue(single_use_txn.HasField("read_write"))
         self.assertEqual(metadata, [("google-cloud-resource-prefix", database.name)])
 
+        self.assertSpanAttributes(
+            "CloudSpanner.Commit", attributes=dict(BASE_ATTRIBUTES, num_mutations=1)
+        )
+
     def test_context_mgr_already_committed(self):
         import datetime
         from google.cloud._helpers import UTC
@@ -276,6 +294,10 @@ def test_context_mgr_success(self):
         self.assertTrue(single_use_txn.HasField("read_write"))
         self.assertEqual(metadata, [("google-cloud-resource-prefix", database.name)])
 
+        self.assertSpanAttributes(
+            "CloudSpanner.Commit", attributes=dict(BASE_ATTRIBUTES, num_mutations=1)
+        )
+
     def test_context_mgr_failure(self):
         import datetime
         from google.cloud.spanner_v1.proto.spanner_pb2 import CommitResponse
diff --git a/tests/unit/test_session.py b/tests/unit/test_session.py
index a39c5e9734..e95b9e1a06 100644
--- a/tests/unit/test_session.py
+++ b/tests/unit/test_session.py
@@ -13,9 +13,13 @@
 # limitations under the License.
 
 
-import unittest
 import google.api_core.gapic_v1.method
 import mock
+from tests._helpers import (
+    OpenTelemetryBase,
+    StatusCanonicalCode,
+    HAS_OPENTELEMETRY_INSTALLED,
+)
 
 
 def _make_rpc_error(error_cls, trailing_metadata=None):
@@ -26,7 +30,12 @@ def _make_rpc_error(error_cls, trailing_metadata=None):
     return error_cls("error", errors=(grpc_error,))
 
 
-class TestSession(unittest.TestCase):
+class _ConstantTime:
+    def time(self):
+        return 1
+
+
+class TestSession(OpenTelemetryBase):
 
     PROJECT_ID = "project-id"
     INSTANCE_ID = "instance-id"
@@ -35,6 +44,12 @@ class TestSession(unittest.TestCase):
     DATABASE_NAME = INSTANCE_NAME + "/databases/" + DATABASE_ID
     SESSION_ID = "session-id"
     SESSION_NAME = DATABASE_NAME + "/sessions/" + SESSION_ID
+    BASE_ATTRIBUTES = {
+        "db.type": "spanner",
+        "db.url": "spanner.googleapis.com:443",
+        "db.instance": DATABASE_NAME,
+        "net.host.name": "spanner.googleapis.com:443",
+    }
 
     def _getTargetClass(self):
         from google.cloud.spanner_v1.session import Session
@@ -107,6 +122,8 @@ def test_create_w_session_id(self):
         with self.assertRaises(ValueError):
             session.create()
 
+        self.assertNoSpans()
+
     def test_create_ok(self):
         session_pb = self._make_session_pb(self.SESSION_NAME)
         gax_api = self._make_spanner_api()
@@ -123,6 +140,10 @@ def test_create_ok(self):
             database.name, metadata=[("google-cloud-resource-prefix", database.name)]
         )
 
+        self.assertSpanAttributes(
+            "CloudSpanner.CreateSession", attributes=TestSession.BASE_ATTRIBUTES
+        )
+
     def test_create_w_labels(self):
         labels = {"foo": "bar"}
         session_pb = self._make_session_pb(self.SESSION_NAME, labels=labels)
@@ -142,6 +163,11 @@ def test_create_w_labels(self):
             metadata=[("google-cloud-resource-prefix", database.name)],
         )
 
+        self.assertSpanAttributes(
+            "CloudSpanner.CreateSession",
+            attributes=dict(TestSession.BASE_ATTRIBUTES, foo="bar"),
+        )
+
     def test_create_error(self):
         from google.api_core.exceptions import Unknown
 
@@ -154,11 +180,19 @@ def test_create_error(self):
         with self.assertRaises(Unknown):
             session.create()
 
+        self.assertSpanAttributes(
+            "CloudSpanner.CreateSession",
+            status=StatusCanonicalCode.UNKNOWN,
+            attributes=TestSession.BASE_ATTRIBUTES,
+        )
+
     def test_exists_wo_session_id(self):
         database = self._make_database()
         session = self._make_one(database)
         self.assertFalse(session.exists())
 
+        self.assertNoSpans()
+
     def test_exists_hit(self):
         session_pb = self._make_session_pb(self.SESSION_NAME)
         gax_api = self._make_spanner_api()
@@ -175,6 +209,11 @@ def test_exists_hit(self):
             metadata=[("google-cloud-resource-prefix", database.name)],
         )
 
+        self.assertSpanAttributes(
+            "CloudSpanner.GetSession",
+            attributes=dict(TestSession.BASE_ATTRIBUTES, session_found=True),
+        )
+
     def test_exists_miss(self):
         from google.api_core.exceptions import NotFound
 
@@ -192,6 +231,11 @@ def test_exists_miss(self):
             metadata=[("google-cloud-resource-prefix", database.name)],
         )
 
+        self.assertSpanAttributes(
+            "CloudSpanner.GetSession",
+            attributes=dict(TestSession.BASE_ATTRIBUTES, session_found=False),
+        )
+
     def test_exists_error(self):
         from google.api_core.exceptions import Unknown
 
@@ -210,6 +254,12 @@ def test_exists_error(self):
             metadata=[("google-cloud-resource-prefix", database.name)],
         )
 
+        self.assertSpanAttributes(
+            "CloudSpanner.GetSession",
+            status=StatusCanonicalCode.UNKNOWN,
+            attributes=TestSession.BASE_ATTRIBUTES,
+        )
+
     def test_ping_wo_session_id(self):
         database = self._make_database()
         session = self._make_one(database)
@@ -277,6 +327,8 @@ def test_delete_wo_session_id(self):
         with self.assertRaises(ValueError):
             session.delete()
 
+        self.assertNoSpans()
+
     def test_delete_hit(self):
         gax_api = self._make_spanner_api()
         gax_api.delete_session.return_value = None
@@ -292,6 +344,10 @@ def test_delete_hit(self):
             metadata=[("google-cloud-resource-prefix", database.name)],
         )
 
+        self.assertSpanAttributes(
+            "CloudSpanner.DeleteSession", attributes=TestSession.BASE_ATTRIBUTES
+        )
+
     def test_delete_miss(self):
         from google.cloud.exceptions import NotFound
 
@@ -310,6 +366,12 @@ def test_delete_miss(self):
             metadata=[("google-cloud-resource-prefix", database.name)],
         )
 
+        self.assertSpanAttributes(
+            "CloudSpanner.DeleteSession",
+            status=StatusCanonicalCode.NOT_FOUND,
+            attributes=TestSession.BASE_ATTRIBUTES,
+        )
+
     def test_delete_error(self):
         from google.api_core.exceptions import Unknown
 
@@ -328,6 +390,12 @@ def test_delete_error(self):
             metadata=[("google-cloud-resource-prefix", database.name)],
         )
 
+        self.assertSpanAttributes(
+            "CloudSpanner.DeleteSession",
+            status=StatusCanonicalCode.UNKNOWN,
+            attributes=TestSession.BASE_ATTRIBUTES,
+        )
+
     def test_snapshot_not_created(self):
         database = self._make_database()
         session = self._make_one(database)
@@ -1044,9 +1112,17 @@ def _time(_results=[1, 1.5]):
             return _results.pop(0)
 
         with mock.patch("time.time", _time):
-            with mock.patch("time.sleep") as sleep_mock:
-                with self.assertRaises(Aborted):
-                    session.run_in_transaction(unit_of_work, "abc", timeout_secs=1)
+            if HAS_OPENTELEMETRY_INSTALLED:
+                with mock.patch("opentelemetry.util.time", _ConstantTime()):
+                    with mock.patch("time.sleep") as sleep_mock:
+                        with self.assertRaises(Aborted):
+                            session.run_in_transaction(
+                                unit_of_work, "abc", timeout_secs=1
+                            )
+            else:
+                with mock.patch("time.sleep") as sleep_mock:
+                    with self.assertRaises(Aborted):
+                        session.run_in_transaction(unit_of_work, "abc", timeout_secs=1)
 
         sleep_mock.assert_not_called()
 
@@ -1106,9 +1182,15 @@ def _time(_results=[1, 2, 4, 8]):
             return _results.pop(0)
 
         with mock.patch("time.time", _time):
-            with mock.patch("time.sleep") as sleep_mock:
-                with self.assertRaises(Aborted):
-                    session.run_in_transaction(unit_of_work, timeout_secs=8)
+            if HAS_OPENTELEMETRY_INSTALLED:
+                with mock.patch("opentelemetry.util.time", _ConstantTime()):
+                    with mock.patch("time.sleep") as sleep_mock:
+                        with self.assertRaises(Aborted):
+                            session.run_in_transaction(unit_of_work, timeout_secs=8)
+            else:
+                with mock.patch("time.sleep") as sleep_mock:
+                    with self.assertRaises(Aborted):
+                        session.run_in_transaction(unit_of_work, timeout_secs=8)
 
         # unpacking call args into list
         call_args = [call_[0][0] for call_ in sleep_mock.call_args_list]
diff --git a/tests/unit/test_snapshot.py b/tests/unit/test_snapshot.py
index 40ba1c6c5a..5c53ee6a0e 100644
--- a/tests/unit/test_snapshot.py
+++ b/tests/unit/test_snapshot.py
@@ -13,10 +13,13 @@
 # limitations under the License.
 
 
-import unittest
 import google.api_core.gapic_v1.method
 import mock
-
+from tests._helpers import (
+    OpenTelemetryBase,
+    StatusCanonicalCode,
+    HAS_OPENTELEMETRY_INSTALLED,
+)
 
 TABLE_NAME = "citizens"
 COLUMNS = ["email", "first_name", "last_name", "age"]
@@ -33,13 +36,19 @@
 TXN_ID = b"DEAFBEAD"
 SECONDS = 3
 MICROS = 123456
+BASE_ATTRIBUTES = {
+    "db.type": "spanner",
+    "db.url": "spanner.googleapis.com:443",
+    "db.instance": "testing",
+    "net.host.name": "spanner.googleapis.com:443",
+}
 
 
-class Test_restart_on_unavailable(unittest.TestCase):
-    def _call_fut(self, restart):
+class Test_restart_on_unavailable(OpenTelemetryBase):
+    def _call_fut(self, restart, span_name=None, session=None, attributes=None):
         from google.cloud.spanner_v1.snapshot import _restart_on_unavailable
 
-        return _restart_on_unavailable(restart)
+        return _restart_on_unavailable(restart, span_name, session, attributes)
 
     def _make_item(self, value, resume_token=b""):
         return mock.Mock(
@@ -51,6 +60,7 @@ def test_iteration_w_empty_raw(self):
         restart = mock.Mock(spec=[], return_value=raw)
         resumable = self._call_fut(restart)
         self.assertEqual(list(resumable), [])
+        self.assertNoSpans()
 
     def test_iteration_w_non_empty_raw(self):
         ITEMS = (self._make_item(0), self._make_item(1))
@@ -59,6 +69,7 @@ def test_iteration_w_non_empty_raw(self):
         resumable = self._call_fut(restart)
         self.assertEqual(list(resumable), list(ITEMS))
         restart.assert_called_once_with()
+        self.assertNoSpans()
 
     def test_iteration_w_raw_w_resume_tken(self):
         ITEMS = (
@@ -72,6 +83,7 @@ def test_iteration_w_raw_w_resume_tken(self):
         resumable = self._call_fut(restart)
         self.assertEqual(list(resumable), list(ITEMS))
         restart.assert_called_once_with()
+        self.assertNoSpans()
 
     def test_iteration_w_raw_raising_unavailable_no_token(self):
         ITEMS = (
@@ -85,6 +97,7 @@ def test_iteration_w_raw_raising_unavailable_no_token(self):
         resumable = self._call_fut(restart)
         self.assertEqual(list(resumable), list(ITEMS))
         self.assertEqual(restart.mock_calls, [mock.call(), mock.call(resume_token=b"")])
+        self.assertNoSpans()
 
     def test_iteration_w_raw_raising_unavailable(self):
         FIRST = (self._make_item(0), self._make_item(1, resume_token=RESUME_TOKEN))
@@ -98,6 +111,7 @@ def test_iteration_w_raw_raising_unavailable(self):
         self.assertEqual(
             restart.mock_calls, [mock.call(), mock.call(resume_token=RESUME_TOKEN)]
         )
+        self.assertNoSpans()
 
     def test_iteration_w_raw_raising_unavailable_after_token(self):
         FIRST = (self._make_item(0), self._make_item(1, resume_token=RESUME_TOKEN))
@@ -110,9 +124,48 @@ def test_iteration_w_raw_raising_unavailable_after_token(self):
         self.assertEqual(
             restart.mock_calls, [mock.call(), mock.call(resume_token=RESUME_TOKEN)]
         )
+        self.assertNoSpans()
 
+    def test_iteration_w_span_creation(self):
+        name = "TestSpan"
+        extra_atts = {"test_att": 1}
+        raw = _MockIterator()
+        restart = mock.Mock(spec=[], return_value=raw)
+        resumable = self._call_fut(restart, name, _Session(_Database()), extra_atts)
+        self.assertEqual(list(resumable), [])
+        self.assertSpanAttributes(name, attributes=dict(BASE_ATTRIBUTES, test_att=1))
+
+    def test_iteration_w_multiple_span_creation(self):
+        if HAS_OPENTELEMETRY_INSTALLED:
+            FIRST = (self._make_item(0), self._make_item(1, resume_token=RESUME_TOKEN))
+            SECOND = (self._make_item(2),)  # discarded after 503
+            LAST = (self._make_item(3),)
+            before = _MockIterator(*(FIRST + SECOND), fail_after=True)
+            after = _MockIterator(*LAST)
+            restart = mock.Mock(spec=[], side_effect=[before, after])
+            name = "TestSpan"
+            resumable = self._call_fut(restart, name, _Session(_Database()))
+            self.assertEqual(list(resumable), list(FIRST + LAST))
+            self.assertEqual(
+                restart.mock_calls, [mock.call(), mock.call(resume_token=RESUME_TOKEN)]
+            )
+
+            span_list = self.memory_exporter.get_finished_spans()
+            self.assertEqual(len(span_list), 2)
+            for span in span_list:
+                self.assertEqual(span.name, name)
+                self.assertEqual(
+                    span.attributes,
+                    {
+                        "db.type": "spanner",
+                        "db.url": "spanner.googleapis.com:443",
+                        "db.instance": "testing",
+                        "net.host.name": "spanner.googleapis.com:443",
+                    },
+                )
 
-class Test_SnapshotBase(unittest.TestCase):
+
+class Test_SnapshotBase(OpenTelemetryBase):
 
     PROJECT_ID = "project-id"
     INSTANCE_ID = "instance-id"
@@ -166,6 +219,8 @@ def test_ctor(self):
         self.assertIs(base._session, session)
         self.assertEqual(base._execute_sql_count, 0)
 
+        self.assertNoSpans()
+
     def test__make_txn_selector_virtual(self):
         session = _Session()
         base = self._make_one(session)
@@ -185,6 +240,14 @@ def test_read_other_error(self):
         with self.assertRaises(RuntimeError):
             list(derived.read(TABLE_NAME, COLUMNS, keyset))
 
+        self.assertSpanAttributes(
+            "CloudSpanner.ReadOnlyTransaction",
+            status=StatusCanonicalCode.UNKNOWN,
+            attributes=dict(
+                BASE_ATTRIBUTES, table_id=TABLE_NAME, columns=tuple(COLUMNS)
+            ),
+        )
+
     def _read_helper(self, multi_use, first=True, count=0, partition=None):
         from google.protobuf.struct_pb2 import Struct
         from google.cloud.spanner_v1.proto.result_set_pb2 import (
@@ -280,6 +343,13 @@ def _read_helper(self, multi_use, first=True, count=0, partition=None):
             metadata=[("google-cloud-resource-prefix", database.name)],
         )
 
+        self.assertSpanAttributes(
+            "CloudSpanner.ReadOnlyTransaction",
+            attributes=dict(
+                BASE_ATTRIBUTES, table_id=TABLE_NAME, columns=tuple(COLUMNS)
+            ),
+        )
+
     def test_read_wo_multi_use(self):
         self._read_helper(multi_use=False)
 
@@ -313,6 +383,12 @@ def test_execute_sql_other_error(self):
 
         self.assertEqual(derived._execute_sql_count, 1)
 
+        self.assertSpanAttributes(
+            "CloudSpanner.ReadWriteTransaction",
+            status=StatusCanonicalCode.UNKNOWN,
+            attributes=dict(BASE_ATTRIBUTES, **{"db.statement": SQL_QUERY}),
+        )
+
     def test_execute_sql_w_params_wo_param_types(self):
         database = _Database()
         session = _Session(database)
@@ -321,6 +397,8 @@ def test_execute_sql_w_params_wo_param_types(self):
         with self.assertRaises(ValueError):
             derived.execute_sql(SQL_QUERY_WITH_PARAM, PARAMS)
 
+        self.assertNoSpans()
+
     def _execute_sql_helper(
         self,
         multi_use,
@@ -440,6 +518,12 @@ def _execute_sql_helper(
 
         self.assertEqual(derived._execute_sql_count, sql_count + 1)
 
+        self.assertSpanAttributes(
+            "CloudSpanner.ReadWriteTransaction",
+            status=StatusCanonicalCode.OK,
+            attributes=dict(BASE_ATTRIBUTES, **{"db.statement": SQL_QUERY_WITH_PARAM}),
+        )
+
     def test_execute_sql_wo_multi_use(self):
         self._execute_sql_helper(multi_use=False)
 
@@ -534,6 +618,14 @@ def _partition_read_helper(
             metadata=[("google-cloud-resource-prefix", database.name)],
         )
 
+        self.assertSpanAttributes(
+            "CloudSpanner.PartitionReadOnlyTransaction",
+            status=StatusCanonicalCode.OK,
+            attributes=dict(
+                BASE_ATTRIBUTES, table_id=TABLE_NAME, columns=tuple(COLUMNS)
+            ),
+        )
+
     def test_partition_read_single_use_raises(self):
         with self.assertRaises(ValueError):
             self._partition_read_helper(multi_use=False, w_txn=True)
@@ -557,6 +649,14 @@ def test_partition_read_other_error(self):
         with self.assertRaises(RuntimeError):
             list(derived.partition_read(TABLE_NAME, COLUMNS, keyset))
 
+        self.assertSpanAttributes(
+            "CloudSpanner.PartitionReadOnlyTransaction",
+            status=StatusCanonicalCode.UNKNOWN,
+            attributes=dict(
+                BASE_ATTRIBUTES, table_id=TABLE_NAME, columns=tuple(COLUMNS)
+            ),
+        )
+
     def test_partition_read_ok_w_index_no_options(self):
         self._partition_read_helper(multi_use=True, w_txn=True, index="index")
 
@@ -626,6 +726,12 @@ def _partition_query_helper(self, multi_use, w_txn, size=None, max_partitions=No
             metadata=[("google-cloud-resource-prefix", database.name)],
         )
 
+        self.assertSpanAttributes(
+            "CloudSpanner.PartitionReadWriteTransaction",
+            status=StatusCanonicalCode.OK,
+            attributes=dict(BASE_ATTRIBUTES, **{"db.statement": SQL_QUERY_WITH_PARAM}),
+        )
+
     def test_partition_query_other_error(self):
         database = _Database()
         database.spanner_api = self._make_spanner_api()
@@ -638,6 +744,12 @@ def test_partition_query_other_error(self):
         with self.assertRaises(RuntimeError):
             list(derived.partition_query(SQL_QUERY))
 
+        self.assertSpanAttributes(
+            "CloudSpanner.PartitionReadWriteTransaction",
+            status=StatusCanonicalCode.UNKNOWN,
+            attributes=dict(BASE_ATTRIBUTES, **{"db.statement": SQL_QUERY}),
+        )
+
     def test_partition_query_w_params_wo_param_types(self):
         database = _Database()
         session = _Session(database)
@@ -648,6 +760,8 @@ def test_partition_query_w_params_wo_param_types(self):
         with self.assertRaises(ValueError):
             list(derived.partition_query(SQL_QUERY_WITH_PARAM, PARAMS))
 
+        self.assertNoSpans()
+
     def test_partition_query_single_use_raises(self):
         with self.assertRaises(ValueError):
             self._partition_query_helper(multi_use=False, w_txn=True)
@@ -666,7 +780,7 @@ def test_partition_query_ok_w_max_partitions(self):
         self._partition_query_helper(multi_use=True, w_txn=True, max_partitions=4)
 
 
-class TestSnapshot(unittest.TestCase):
+class TestSnapshot(OpenTelemetryBase):
 
     PROJECT_ID = "project-id"
     INSTANCE_ID = "instance-id"
@@ -933,6 +1047,12 @@ def test_begin_w_other_error(self):
         with self.assertRaises(RuntimeError):
             snapshot.begin()
 
+        self.assertSpanAttributes(
+            "CloudSpanner.BeginTransaction",
+            status=StatusCanonicalCode.UNKNOWN,
+            attributes=BASE_ATTRIBUTES,
+        )
+
     def test_begin_ok_exact_staleness(self):
         from google.protobuf.duration_pb2 import Duration
         from google.cloud.spanner_v1.proto.transaction_pb2 import (
@@ -964,6 +1084,12 @@ def test_begin_ok_exact_staleness(self):
             metadata=[("google-cloud-resource-prefix", database.name)],
         )
 
+        self.assertSpanAttributes(
+            "CloudSpanner.BeginTransaction",
+            status=StatusCanonicalCode.OK,
+            attributes=BASE_ATTRIBUTES,
+        )
+
     def test_begin_ok_exact_strong(self):
         from google.cloud.spanner_v1.proto.transaction_pb2 import (
             Transaction as TransactionPB,
@@ -992,6 +1118,12 @@ def test_begin_ok_exact_strong(self):
             metadata=[("google-cloud-resource-prefix", database.name)],
         )
 
+        self.assertSpanAttributes(
+            "CloudSpanner.BeginTransaction",
+            status=StatusCanonicalCode.OK,
+            attributes=BASE_ATTRIBUTES,
+        )
+
 
 class _Client(object):
     def __init__(self):
diff --git a/tests/unit/test_transaction.py b/tests/unit/test_transaction.py
index 6ae24aedab..e2ac7c2eec 100644
--- a/tests/unit/test_transaction.py
+++ b/tests/unit/test_transaction.py
@@ -13,10 +13,8 @@
 # limitations under the License.
 
 
-import unittest
-
 import mock
-
+from tests._helpers import OpenTelemetryBase, StatusCanonicalCode
 
 TABLE_NAME = "citizens"
 COLUMNS = ["email", "first_name", "last_name", "age"]
@@ -36,7 +34,7 @@
 PARAM_TYPES = {"age": "INT64"}
 
 
-class TestTransaction(unittest.TestCase):
+class TestTransaction(OpenTelemetryBase):
 
     PROJECT_ID = "project-id"
     INSTANCE_ID = "instance-id"
@@ -47,6 +45,13 @@ class TestTransaction(unittest.TestCase):
     SESSION_NAME = DATABASE_NAME + "/sessions/" + SESSION_ID
     TRANSACTION_ID = b"DEADBEEF"
 
+    BASE_ATTRIBUTES = {
+        "db.type": "spanner",
+        "db.url": "spanner.googleapis.com:443",
+        "db.instance": "testing",
+        "net.host.name": "spanner.googleapis.com:443",
+    }
+
     def _getTargetClass(self):
         from google.cloud.spanner_v1.transaction import Transaction
 
@@ -122,6 +127,8 @@ def test_begin_already_begun(self):
         with self.assertRaises(ValueError):
             transaction.begin()
 
+        self.assertNoSpans()
+
     def test_begin_already_rolled_back(self):
         session = _Session()
         transaction = self._make_one(session)
@@ -129,6 +136,8 @@ def test_begin_already_rolled_back(self):
         with self.assertRaises(ValueError):
             transaction.begin()
 
+        self.assertNoSpans()
+
     def test_begin_already_committed(self):
         session = _Session()
         transaction = self._make_one(session)
@@ -136,6 +145,8 @@ def test_begin_already_committed(self):
         with self.assertRaises(ValueError):
             transaction.begin()
 
+        self.assertNoSpans()
+
     def test_begin_w_other_error(self):
         database = _Database()
         database.spanner_api = self._make_spanner_api()
@@ -146,6 +157,12 @@ def test_begin_w_other_error(self):
         with self.assertRaises(RuntimeError):
             transaction.begin()
 
+        self.assertSpanAttributes(
+            "CloudSpanner.BeginTransaction",
+            status=StatusCanonicalCode.UNKNOWN,
+            attributes=TestTransaction.BASE_ATTRIBUTES,
+        )
+
     def test_begin_ok(self):
         from google.cloud.spanner_v1.proto.transaction_pb2 import (
             Transaction as TransactionPB,
@@ -169,12 +186,18 @@ def test_begin_ok(self):
         self.assertTrue(txn_options.HasField("read_write"))
         self.assertEqual(metadata, [("google-cloud-resource-prefix", database.name)])
 
+        self.assertSpanAttributes(
+            "CloudSpanner.BeginTransaction", attributes=TestTransaction.BASE_ATTRIBUTES
+        )
+
     def test_rollback_not_begun(self):
         session = _Session()
         transaction = self._make_one(session)
         with self.assertRaises(ValueError):
             transaction.rollback()
 
+        self.assertNoSpans()
+
     def test_rollback_already_committed(self):
         session = _Session()
         transaction = self._make_one(session)
@@ -183,6 +206,8 @@ def test_rollback_already_committed(self):
         with self.assertRaises(ValueError):
             transaction.rollback()
 
+        self.assertNoSpans()
+
     def test_rollback_already_rolled_back(self):
         session = _Session()
         transaction = self._make_one(session)
@@ -191,6 +216,8 @@ def test_rollback_already_rolled_back(self):
         with self.assertRaises(ValueError):
             transaction.rollback()
 
+        self.assertNoSpans()
+
     def test_rollback_w_other_error(self):
         database = _Database()
         database.spanner_api = self._make_spanner_api()
@@ -205,6 +232,12 @@ def test_rollback_w_other_error(self):
 
         self.assertFalse(transaction.rolled_back)
 
+        self.assertSpanAttributes(
+            "CloudSpanner.Rollback",
+            status=StatusCanonicalCode.UNKNOWN,
+            attributes=TestTransaction.BASE_ATTRIBUTES,
+        )
+
     def test_rollback_ok(self):
         from google.protobuf.empty_pb2 import Empty
 
@@ -226,12 +259,18 @@ def test_rollback_ok(self):
         self.assertEqual(txn_id, self.TRANSACTION_ID)
         self.assertEqual(metadata, [("google-cloud-resource-prefix", database.name)])
 
+        self.assertSpanAttributes(
+            "CloudSpanner.Rollback", attributes=TestTransaction.BASE_ATTRIBUTES
+        )
+
     def test_commit_not_begun(self):
         session = _Session()
         transaction = self._make_one(session)
         with self.assertRaises(ValueError):
             transaction.commit()
 
+        self.assertNoSpans()
+
     def test_commit_already_committed(self):
         session = _Session()
         transaction = self._make_one(session)
@@ -240,6 +279,8 @@ def test_commit_already_committed(self):
         with self.assertRaises(ValueError):
             transaction.commit()
 
+        self.assertNoSpans()
+
     def test_commit_already_rolled_back(self):
         session = _Session()
         transaction = self._make_one(session)
@@ -248,6 +289,8 @@ def test_commit_already_rolled_back(self):
         with self.assertRaises(ValueError):
             transaction.commit()
 
+        self.assertNoSpans()
+
     def test_commit_w_other_error(self):
         database = _Database()
         database.spanner_api = self._make_spanner_api()
@@ -262,6 +305,12 @@ def test_commit_w_other_error(self):
 
         self.assertIsNone(transaction.committed)
 
+        self.assertSpanAttributes(
+            "CloudSpanner.Commit",
+            status=StatusCanonicalCode.UNKNOWN,
+            attributes=dict(TestTransaction.BASE_ATTRIBUTES, num_mutations=1),
+        )
+
     def _commit_helper(self, mutate=True):
         import datetime
         from google.cloud.spanner_v1.proto.spanner_pb2 import CommitResponse
@@ -294,6 +343,14 @@ def _commit_helper(self, mutate=True):
         self.assertEqual(mutations, transaction._mutations)
         self.assertEqual(metadata, [("google-cloud-resource-prefix", database.name)])
 
+        self.assertSpanAttributes(
+            "CloudSpanner.Commit",
+            attributes=dict(
+                TestTransaction.BASE_ATTRIBUTES,
+                num_mutations=len(transaction._mutations),
+            ),
+        )
+
     def test_commit_no_mutations(self):
         self._commit_helper(mutate=False)