feat(opentelemetry): add request_tag to span attributes

Author: sinhasubham

examples/trace.py

@@ -27,7 +27,7 @@
 from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
 
 # Setup common variables that'll be used between Spanner and traces.
-project_id = os.environ.get('SPANNER_PROJECT_ID', 'span-cloud-testing')
+project_id = os.environ.get('SPANNER_PROJECT_ID', 'test-project')
 
 def spanner_with_cloud_trace():
     # [START spanner_opentelemetry_traces_cloudtrace_usage]
@@ -62,13 +62,16 @@ def spanner_with_otlp():
 
 
 def main():
+    # Setup OpenTelemetry, trace and Cloud Trace exporter.
+    tracer_provider = TracerProvider(sampler=ALWAYS_ON)
+    trace_exporter = CloudTraceSpanExporter(project_id=project_id)
+    tracer_provider.add_span_processor(BatchSpanProcessor(trace_exporter))
+
     # Setup the Cloud Spanner Client.
     # Change to "spanner_client = spanner_with_otlp" to use OTLP exporter
     spanner_client = spanner_with_cloud_trace()
-    instance = spanner_client.instance('suvham-testing')
-    instance.reload()
-    database = instance.database('gildb')
-    tracer_provider = spanner_client.observability_options["tracer_provider"]
+    instance = spanner_client.instance('test-instance')
+    database = instance.database('test-db')
 
     # Set W3C Trace Context as the global propagator for end to end tracing.
     set_global_textmap(TraceContextTextMapPropagator())
@@ -90,27 +93,12 @@ def main():
                 # Purposefully issue a bad SQL statement to examine exceptions
                 # that get recorded and a ERROR span status.
                 try:
-                    data = snapshot.execute_sql('SELECT CURRENT_TIMESTAMP()')
+                    data = snapshot.execute_sql('SELECT CURRENT_TIMESTAMPx()')
                     for row in data:
                         print(row)
                 except Exception as e:
                     print(e)
 
-        # Example of a read-write transaction with a transaction tag
-        with tracer.start_as_current_span('TaggedTransaction'):
-            def update_singer_name(transaction):
-                transaction.execute_update(
-                    "UPDATE Singers SET FirstName = 'Timothy' WHERE SingerId = 1",
-                    request_options={
-                        "request_tag": "app=concert,env=dev,action=update"
-                    },
-                )
-                print("Updated singer's name.")
-
-            database.run_in_transaction(
-                update_singer_name, transaction_tag="app=concert,env=dev"
-            )
-
 
 if __name__ == '__main__':
     main()

google/cloud/spanner_v1/_opentelemetry_tracing.py

@@ -122,6 +122,11 @@ def trace_call(
     if extra_attributes:
         attributes.update(extra_attributes)
 
+    if "request_options" in attributes:
+        request_options = attributes.pop("request_options")
+        if request_options and request_options.request_tag:
+            attributes["spanner.request_tag"] = request_options.request_tag
+
     if extended_tracing_globally_disabled:
         enable_extended_tracing = False
 

google/cloud/spanner_v1/snapshot.py

@@ -409,7 +409,11 @@ def read(
             method=streaming_read_method,
             request=read_request,
             metadata=metadata,
-            trace_attributes={"table_id": table, "columns": columns},
+            trace_attributes={
+                "table_id": table,
+                "columns": columns,
+                "request_options": request_options,
+            },
             column_info=column_info,
             lazy_decode=lazy_decode,
         )
@@ -601,7 +605,7 @@ def execute_sql(
             method=execute_streaming_sql_method,
             request=execute_sql_request,
             metadata=metadata,
-            trace_attributes={"db.statement": sql},
+            trace_attributes={"db.statement": sql, "request_options": request_options},
             column_info=column_info,
             lazy_decode=lazy_decode,
         )

tests/unit/test_snapshot.py

@@ -116,6 +116,8 @@
 class _Derived(_SnapshotBase):
     """A minimally-implemented _SnapshotBase-derived class for testing"""
 
+    transaction_tag = None
+
     # Use a simplified implementation of _build_transaction_options_pb
     # that always returns the same transaction options.
     TRANSACTION_OPTIONS = TransactionOptions()
@@ -619,23 +621,23 @@ def test_iteration_w_multiple_span_creation(self):
 
 
 class Test_SnapshotBase(OpenTelemetryBase):
-    # def test_ctor(self):
-    #     session = build_session()
-    #     derived = _build_snapshot_derived(session=session)
-    #
-    #     # Attributes from _SessionWrapper.
-    #     self.assertIs(derived._session, session)
-    #
-    #     # Attributes from _SnapshotBase.
-    #     self.assertTrue(derived._read_only)
-    #     self.assertFalse(derived._multi_use)
-    #     self.assertEqual(derived._execute_sql_request_count, 0)
-    #     self.assertEqual(derived._read_request_count, 0)
-    #     self.assertIsNone(derived._transaction_id)
-    #     self.assertIsNone(derived._precommit_token)
-    #     self.assertIsInstance(derived._lock, type(Lock()))
-    #
-    #     self.assertNoSpans()
+    def test_ctor(self):
+        session = build_session()
+        derived = _build_snapshot_derived(session=session)
+    
+        # Attributes from _SessionWrapper.
+        self.assertIs(derived._session, session)
+    
+        # Attributes from _SnapshotBase.
+        self.assertTrue(derived._read_only)
+        self.assertFalse(derived._multi_use)
+        self.assertEqual(derived._execute_sql_request_count, 0)
+        self.assertEqual(derived._read_request_count, 0)
+        self.assertIsNone(derived._transaction_id)
+        self.assertIsNone(derived._precommit_token)
+        self.assertIsInstance(derived._lock, type(Lock()))
+    
+        self.assertNoSpans()
 
     def test__build_transaction_selector_pb_single_use(self):
         derived = _build_snapshot_derived(multi_use=False)
@@ -1282,7 +1284,10 @@ def _execute_sql_helper(
 
         expected_attributes = dict(
             BASE_ATTRIBUTES,
-            **{"db.statement": SQL_QUERY_WITH_PARAM, "x_goog_spanner_request_id": req_id},
+            **{
+                "db.statement": SQL_QUERY_WITH_PARAM,
+                "x_goog_spanner_request_id": req_id,
+            },
         )
         if request_options and request_options.request_tag:
             expected_attributes["spanner.request_tag"] = request_options.request_tag

tests/unit/test_spanner_metrics_tracer_factory.py

@@ -14,12 +14,9 @@
 # limitations under the License.
 
 import pytest
-import unittest
-from unittest import mock
 from google.cloud.spanner_v1.metrics.spanner_metrics_tracer_factory import (
     SpannerMetricsTracerFactory,
 )
-from opentelemetry.sdk.resources import Resource
 
 pytest.importorskip("opentelemetry")