Skip to content

Commit 87341e3

Browse files
odeke-emodeke-em
committed
.github/workflows: toggle and test with EXTENDED_TRACING_ENABLED=true
1 parent 9a315b9 commit 87341e3

File tree

5 files changed

+106
-75
lines changed

5 files changed

+106
-75
lines changed

‎.github/workflows/integration-tests-against-emulator.yaml

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,9 +24,17 @@ jobs:
2424
python-version: 3.8
2525
- name: Install nox
2626
run: python -m pip install nox
27-
- name: Run system tests
27+
- name: Run system tests (without extended tracing)
2828
run: nox -s system
2929
env:
3030
SPANNER_EMULATOR_HOST: localhost:9010
3131
GOOGLE_CLOUD_PROJECT: emulator-test-project
3232
GOOGLE_CLOUD_TESTS_CREATE_SPANNER_INSTANCE: true
33+
EXTENDED_TRACING_ENABLED: false
34+
- name: Run system tests (with extended tracing)
35+
run: nox -s system
36+
env:
37+
SPANNER_EMULATOR_HOST: localhost:9010
38+
GOOGLE_CLOUD_PROJECT: emulator-test-project
39+
GOOGLE_CLOUD_TESTS_CREATE_SPANNER_INSTANCE: true
40+
EXTENDED_TRACING_ENABLED: true

‎google/cloud/spanner_v1/transaction.py

Lines changed: 50 additions & 57 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@
1111
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
1212
# See the License for the specific language governing permissions and
1313
# limitations under the License.
14+
1415
"""Spanner read-write transaction support."""
1516
import functools
1617
import threading
@@ -90,20 +91,18 @@ def _make_txn_selector(self):
9091
self._check_state()
9192

9293
if self._transaction_id is None:
93-
return TransactionSelector(begin=TransactionOptions(
94-
read_write=TransactionOptions.ReadWrite(),
95-
exclude_txn_from_change_streams=self.
96-
exclude_txn_from_change_streams,
97-
))
94+
return TransactionSelector(
95+
begin=TransactionOptions(
96+
read_write=TransactionOptions.ReadWrite(),
97+
exclude_txn_from_change_streams=self.exclude_txn_from_change_streams,
98+
)
99+
)
98100
else:
99101
return TransactionSelector(id=self._transaction_id)
100102

101-
def _execute_request(self,
102-
method,
103-
request,
104-
trace_name=None,
105-
session=None,
106-
attributes=None):
103+
def _execute_request(
104+
self, method, request, trace_name=None, session=None, attributes=None
105+
):
107106
"""Helper method to execute request after fetching transaction selector.
108107
109108
:type method: callable
@@ -118,9 +117,7 @@ def _execute_request(self,
118117
method = functools.partial(method, request=request)
119118
response = _retry(
120119
method,
121-
allowed_exceptions={
122-
InternalServerError: _check_rst_stream_error
123-
},
120+
allowed_exceptions={InternalServerError: _check_rst_stream_error},
124121
)
125122

126123
return response
@@ -147,12 +144,11 @@ def begin(self):
147144
metadata = _metadata_with_prefix(database.name)
148145
if database._route_to_leader_enabled:
149146
metadata.append(
150-
_metadata_with_leader_aware_routing(
151-
database._route_to_leader_enabled))
147+
_metadata_with_leader_aware_routing(database._route_to_leader_enabled)
148+
)
152149
txn_options = TransactionOptions(
153150
read_write=TransactionOptions.ReadWrite(),
154-
exclude_txn_from_change_streams=self.
155-
exclude_txn_from_change_streams,
151+
exclude_txn_from_change_streams=self.exclude_txn_from_change_streams,
156152
)
157153
with trace_call("CloudSpanner.BeginTransaction", self._session):
158154
method = functools.partial(
@@ -163,9 +159,7 @@ def begin(self):
163159
)
164160
response = _retry(
165161
method,
166-
allowed_exceptions={
167-
InternalServerError: _check_rst_stream_error
168-
},
162+
allowed_exceptions={InternalServerError: _check_rst_stream_error},
169163
)
170164
self._transaction_id = response.id
171165
return self._transaction_id
@@ -181,7 +175,9 @@ def rollback(self):
181175
if database._route_to_leader_enabled:
182176
metadata.append(
183177
_metadata_with_leader_aware_routing(
184-
database._route_to_leader_enabled))
178+
database._route_to_leader_enabled
179+
)
180+
)
185181
with trace_call("CloudSpanner.Rollback", self._session):
186182
method = functools.partial(
187183
api.rollback,
@@ -191,17 +187,14 @@ def rollback(self):
191187
)
192188
_retry(
193189
method,
194-
allowed_exceptions={
195-
InternalServerError: _check_rst_stream_error
196-
},
190+
allowed_exceptions={InternalServerError: _check_rst_stream_error},
197191
)
198192
self.rolled_back = True
199193
del self._session._transaction
200194

201-
def commit(self,
202-
return_commit_stats=False,
203-
request_options=None,
204-
max_commit_delay=None):
195+
def commit(
196+
self, return_commit_stats=False, request_options=None, max_commit_delay=None
197+
):
205198
"""Commit mutations to the database.
206199
207200
:type return_commit_stats: bool
@@ -236,8 +229,8 @@ def commit(self,
236229
metadata = _metadata_with_prefix(database.name)
237230
if database._route_to_leader_enabled:
238231
metadata.append(
239-
_metadata_with_leader_aware_routing(
240-
database._route_to_leader_enabled))
232+
_metadata_with_leader_aware_routing(database._route_to_leader_enabled)
233+
)
241234
trace_attributes = {"num_mutations": len(self._mutations)}
242235

243236
if request_options is None:
@@ -258,18 +251,15 @@ def commit(self,
258251
max_commit_delay=max_commit_delay,
259252
request_options=request_options,
260253
)
261-
with trace_call("CloudSpanner.Commit", self._session,
262-
trace_attributes):
254+
with trace_call("CloudSpanner.Commit", self._session, trace_attributes):
263255
method = functools.partial(
264256
api.commit,
265257
request=request,
266258
metadata=metadata,
267259
)
268260
response = _retry(
269261
method,
270-
allowed_exceptions={
271-
InternalServerError: _check_rst_stream_error
272-
},
262+
allowed_exceptions={InternalServerError: _check_rst_stream_error},
273263
)
274264
self.committed = response.commit_timestamp
275265
if return_commit_stats:
@@ -298,10 +288,9 @@ def _make_params_pb(params, param_types):
298288
If ``params`` is None but ``param_types`` is not None.
299289
"""
300290
if params is not None:
301-
return Struct(fields={
302-
key: _make_value_pb(value)
303-
for key, value in params.items()
304-
})
291+
return Struct(
292+
fields={key: _make_value_pb(value) for key, value in params.items()}
293+
)
305294

306295
return {}
307296

@@ -363,8 +352,8 @@ def execute_update(
363352
metadata = _metadata_with_prefix(database.name)
364353
if database._route_to_leader_enabled:
365354
metadata.append(
366-
_metadata_with_leader_aware_routing(
367-
database._route_to_leader_enabled))
355+
_metadata_with_leader_aware_routing(database._route_to_leader_enabled)
356+
)
368357
api = database.spanner_api
369358

370359
seqno, self._execute_sql_count = (
@@ -375,8 +364,7 @@ def execute_update(
375364
# Query-level options have higher precedence than client-level and
376365
# environment-level options
377366
default_query_options = database._instance._client._query_options
378-
query_options = _merge_query_options(default_query_options,
379-
query_options)
367+
query_options = _merge_query_options(default_query_options, query_options)
380368

381369
if request_options is None:
382370
request_options = RequestOptions()
@@ -416,9 +404,12 @@ def execute_update(
416404
trace_attributes,
417405
)
418406
# Setting the transaction id because the transaction begin was inlined for first rpc.
419-
if (self._transaction_id is None and response is not None
420-
and response.metadata is not None
421-
and response.metadata.transaction is not None):
407+
if (
408+
self._transaction_id is None
409+
and response is not None
410+
and response.metadata is not None
411+
and response.metadata.transaction is not None
412+
):
422413
self._transaction_id = response.metadata.transaction.id
423414
else:
424415
response = self._execute_request(
@@ -481,16 +472,17 @@ def batch_update(
481472
dml, params, param_types = statement
482473
params_pb = self._make_params_pb(params, param_types)
483474
parsed.append(
484-
ExecuteBatchDmlRequest.Statement(sql=dml,
485-
params=params_pb,
486-
param_types=param_types))
475+
ExecuteBatchDmlRequest.Statement(
476+
sql=dml, params=params_pb, param_types=param_types
477+
)
478+
)
487479

488480
database = self._session._database
489481
metadata = _metadata_with_prefix(database.name)
490482
if database._route_to_leader_enabled:
491483
metadata.append(
492-
_metadata_with_leader_aware_routing(
493-
database._route_to_leader_enabled))
484+
_metadata_with_leader_aware_routing(database._route_to_leader_enabled)
485+
)
494486
api = database.spanner_api
495487

496488
seqno, self._execute_sql_count = (
@@ -535,9 +527,11 @@ def batch_update(
535527
)
536528
# Setting the transaction id because the transaction begin was inlined for first rpc.
537529
for result_set in response.result_sets:
538-
if (self._transaction_id is None
539-
and result_set.metadata is not None
540-
and result_set.metadata.transaction is not None):
530+
if (
531+
self._transaction_id is None
532+
and result_set.metadata is not None
533+
and result_set.metadata.transaction is not None
534+
):
541535
self._transaction_id = result_set.metadata.transaction.id
542536
break
543537
else:
@@ -550,8 +544,7 @@ def batch_update(
550544
)
551545

552546
row_counts = [
553-
result_set.stats.row_count_exact
554-
for result_set in response.result_sets
547+
result_set.stats.row_count_exact for result_set in response.result_sets
555548
]
556549

557550
return response.status, row_counts

‎tests/_helpers.py

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
import os
12
import unittest
23
import mock
34

@@ -32,6 +33,7 @@
3233

3334
_TEST_OT_EXPORTER = None
3435
_TEST_OT_PROVIDER_INITIALIZED = False
36+
EXTENDED_TRACING_ENABLED = os.environ.get('SPANNER_ENABLE_EXTENDED_TRACING', '') == 'true'
3537

3638

3739
def get_test_ot_exporter():

‎tests/unit/test_session.py

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,7 @@ class TestSession(OpenTelemetryBase):
5555
DB_NAME: DATABASE_NAME,
5656
NET_HOST_NAME: "spanner.googleapis.com",
5757
}
58+
5859
def _getTargetClass(self):
5960
from google.cloud.spanner_v1.session import Session
6061

‎tests/unit/test_snapshot.py

Lines changed: 44 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -20,12 +20,12 @@
2020
from tests._helpers import (
2121
OpenTelemetryBase,
2222
StatusCode,
23+
EXTENDED_TRACING_ENABLED,
2324
HAS_OPENTELEMETRY_INSTALLED,
2425
DB_STATEMENT,
2526
DB_SYSTEM,
2627
DB_NAME,
2728
DB_CONNECTION_STRING,
28-
EXTENDED_TRACING_EANBLED,
2929
NET_HOST_NAME,
3030
)
3131
from google.cloud.spanner_v1.param_types import INT64
@@ -868,12 +868,18 @@ def test_execute_sql_other_error(self):
868868

869869
self.assertEqual(derived._execute_sql_count, 1)
870870

871-
if EXTENDED_TRACING_EANBLED:
871+
if EXTENDED_TRACING_ENABLED:
872872
self.assertSpanAttributes(
873873
"CloudSpanner.ReadWriteTransaction",
874874
status=StatusCode.ERROR,
875875
attributes=dict(BASE_ATTRIBUTES, **{DB_STATEMENT: SQL_QUERY}),
876876
)
877+
else:
878+
self.assertSpanAttributes(
879+
"CloudSpanner.ReadWriteTransaction",
880+
status=StatusCode.ERROR,
881+
attributes=BASE_ATTRIBUTES,
882+
)
877883

878884
def _execute_sql_helper(
879885
self,
@@ -1025,11 +1031,18 @@ def _execute_sql_helper(
10251031

10261032
self.assertEqual(derived._execute_sql_count, sql_count + 1)
10271033

1028-
self.assertSpanAttributes(
1029-
"CloudSpanner.ReadWriteTransaction",
1030-
status=StatusCode.OK,
1031-
attributes=dict(BASE_ATTRIBUTES, **{DB_STATEMENT: SQL_QUERY_WITH_PARAM}),
1032-
)
1034+
if EXTENDED_TRACING_ENABLED:
1035+
self.assertSpanAttributes(
1036+
"CloudSpanner.ReadWriteTransaction",
1037+
status=StatusCode.OK,
1038+
attributes=dict(BASE_ATTRIBUTES, **{DB_STATEMENT: SQL_QUERY_WITH_PARAM}),
1039+
)
1040+
else:
1041+
self.assertSpanAttributes(
1042+
"CloudSpanner.ReadWriteTransaction",
1043+
status=StatusCode.OK,
1044+
attributes=BASE_ATTRIBUTES,
1045+
)
10331046

10341047
def test_execute_sql_wo_multi_use(self):
10351048
self._execute_sql_helper(multi_use=False)
@@ -1370,11 +1383,18 @@ def _partition_query_helper(
13701383
timeout=timeout,
13711384
)
13721385

1373-
self.assertSpanAttributes(
1374-
"CloudSpanner.PartitionReadWriteTransaction",
1375-
status=StatusCode.OK,
1376-
attributes=dict(BASE_ATTRIBUTES, **{DB_STATEMENT: SQL_QUERY_WITH_PARAM}),
1377-
)
1386+
if EXTENDED_TRACING_ENABLED:
1387+
self.assertSpanAttributes(
1388+
"CloudSpanner.PartitionReadWriteTransaction",
1389+
status=StatusCode.OK,
1390+
attributes=dict(BASE_ATTRIBUTES, **{DB_STATEMENT: SQL_QUERY_WITH_PARAM}),
1391+
)
1392+
else:
1393+
self.assertSpanAttributes(
1394+
"CloudSpanner.PartitionReadWriteTransaction",
1395+
status=StatusCode.OK,
1396+
attributes=BASE_ATTRIBUTES,
1397+
)
13781398

13791399
def test_partition_query_other_error(self):
13801400
database = _Database()
@@ -1388,11 +1408,18 @@ def test_partition_query_other_error(self):
13881408
with self.assertRaises(RuntimeError):
13891409
list(derived.partition_query(SQL_QUERY))
13901410

1391-
self.assertSpanAttributes(
1392-
"CloudSpanner.PartitionReadWriteTransaction",
1393-
status=StatusCode.ERROR,
1394-
attributes=dict(BASE_ATTRIBUTES, **{DB_STATEMENT: SQL_QUERY}),
1395-
)
1411+
if EXTENDED_TRACING_ENABLED:
1412+
self.assertSpanAttributes(
1413+
"CloudSpanner.PartitionReadWriteTransaction",
1414+
status=StatusCode.ERROR,
1415+
attributes=dict(BASE_ATTRIBUTES, **{DB_STATEMENT: SQL_QUERY}),
1416+
)
1417+
else:
1418+
self.assertSpanAttributes(
1419+
"CloudSpanner.PartitionReadWriteTransaction",
1420+
status=StatusCode.ERROR,
1421+
attributes=BASE_ATTRIBUTES,
1422+
)
13961423

13971424
def test_partition_query_single_use_raises(self):
13981425
with self.assertRaises(ValueError):

0 commit comments

Comments
 (0)