feat: add span attributes containing request and transaction tags

Signed-off-by: Sri Harsha CH <sriharshach@google.com>

Author: harshachinta

google/cloud/spanner_v1/batch.py

@@ -222,7 +222,10 @@ def commit(
             database.default_transaction_options.default_read_write_transaction_options,
             txn_options,
         )
-        trace_attributes = {"num_mutations": len(self._mutations)}
+        trace_attributes = {
+            "num_mutations": len(self._mutations),
+            "transaction.tag": self.transaction_tag,
+        }
 
         if request_options is None:
             request_options = RequestOptions()
@@ -352,12 +355,15 @@ def batch_write(self, request_options=None, exclude_txn_from_change_streams=Fals
             metadata.append(
                 _metadata_with_leader_aware_routing(database._route_to_leader_enabled)
             )
-        trace_attributes = {"num_mutation_groups": len(self._mutation_groups)}
         if request_options is None:
             request_options = RequestOptions()
         elif type(request_options) is dict:
             request_options = RequestOptions(request_options)
 
+        trace_attributes = {
+            "num_mutation_groups": len(self._mutation_groups),
+            "request.tag": request_options.request_tag,
+        }
         request = BatchWriteRequest(
             session=self._session.name,
             mutation_groups=self._mutation_groups,

google/cloud/spanner_v1/session.py

@@ -472,9 +472,11 @@ def run_in_transaction(self, func, *args, **kw):
         attempts = 0
 
         observability_options = getattr(self._database, "observability_options", None)
+        trace_attributes = {"transaction.tag": transaction_tag}
         with trace_call(
             "CloudSpanner.Session.run_in_transaction",
             self,
+            extra_attributes=trace_attributes,
             observability_options=observability_options,
         ) as span, MetricsCapture():
             while True:

google/cloud/spanner_v1/snapshot.py

@@ -337,7 +337,12 @@ def read(
             timeout=timeout,
         )
 
-        trace_attributes = {"table_id": table, "columns": columns}
+        trace_attributes = {
+            "table_id": table,
+            "columns": columns,
+            "request.tag": request_options.request_tag,
+            "transaction.tag": request_options.transaction_tag,
+        }
         observability_options = getattr(database, "observability_options", None)
 
         if self._transaction_id is None:
@@ -570,7 +575,11 @@ def execute_sql(
             timeout=timeout,
         )
 
-        trace_attributes = {"db.statement": sql}
+        trace_attributes = {
+            "db.statement": sql,
+            "request.tag": request_options.request_tag,
+            "transaction.tag": request_options.transaction_tag,
+        }
         observability_options = getattr(database, "observability_options", None)
 
         if self._transaction_id is None:

google/cloud/spanner_v1/transaction.py

@@ -264,7 +264,10 @@ def commit(
         :raises ValueError: if there are no mutations to commit.
         """
         database = self._session._database
-        trace_attributes = {"num_mutations": len(self._mutations)}
+        trace_attributes = {
+            "num_mutations": len(self._mutations),
+            "transaction.tag": self.transaction_tag,
+        }
         observability_options = getattr(database, "observability_options", None)
         api = database.spanner_api
         metadata = _metadata_with_prefix(database.name)
@@ -455,7 +458,11 @@ def execute_update(
             request_options = RequestOptions(request_options)
         request_options.transaction_tag = self.transaction_tag
 
-        trace_attributes = {"db.statement": dml}
+        trace_attributes = {
+            "db.statement": dml,
+            "request.tag": request_options.request_tag,
+            "transaction.tag": request_options.transaction_tag,
+        }
 
         request = ExecuteSqlRequest(
             session=self._session.name,
@@ -601,7 +608,9 @@ def batch_update(
 
         trace_attributes = {
             # Get just the queries from the DML statement batch
-            "db.statement": ";".join([statement.sql for statement in parsed])
+            "db.statement": ";".join([statement.sql for statement in parsed]),
+            "request.tag": request_options.request_tag,
+            "transaction.tag": request_options.transaction_tag,
         }
         request = ExecuteBatchDmlRequest(
             session=self._session.name,