Skip to content
Navigation Menu
{{ message }}
-
Notifications
You must be signed in to change notification settings - Fork 1.7k
feat(bigtable): add client side metric instrumentation to basic rpcs #16712
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Merged
Merged
Changes from all commits
Commits
Show all changes
11 commits
Select commit
Hold shift + click to select a range
8a3af45
moved over instrumentation code
daniel-sanche cf58c57
refactored system tests
daniel-sanche 00618f8
updated tests
daniel-sanche ead3bf1
fixed lint
daniel-sanche af0beba
Merge branch 'main' into bigtable_csm_1_basic_instrumentation
daniel-sanche d0c03d0
regenerated files
daniel-sanche 01f1739
fixed mtls tests
daniel-sanche d40fa0a
fixed format
daniel-sanche febdaac
Update packages/google-cloud-bigtable/tests/system/data/test_metrics_…
daniel-sanche c7b142d
improve system test cleanup
daniel-sanche e936535
fix lint
daniel-sanche File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -63,7 +63,11 @@ | |
| _validate_timeouts, | ||
| _WarmedInstanceKey, | ||
| ) | ||
| from google.cloud.bigtable.data._metrics import BigtableClientSideMetricsController | ||
| from google.cloud.bigtable.data._metrics import ( | ||
| BigtableClientSideMetricsController, | ||
| OperationType, | ||
| tracked_retry, | ||
| ) | ||
| from google.cloud.bigtable.data.exceptions import ( | ||
| FailedQueryShardError, | ||
| ShardedReadRowsExceptionGroup, | ||
|
|
@@ -1431,26 +1435,28 @@ async def sample_row_keys( | |
| retryable_excs = _get_retryable_errors(retryable_errors, self) | ||
| predicate = retries.if_exception_type(*retryable_excs) | ||
|
|
||
| sleep_generator = retries.exponential_sleep_generator(0.01, 2, 60) | ||
|
|
||
| @CrossSync.convert | ||
| async def execute_rpc(): | ||
| results = await self.client._gapic_client.sample_row_keys( | ||
| request=SampleRowKeysRequest( | ||
| app_profile_id=self.app_profile_id, **self._request_path | ||
| ), | ||
| timeout=next(attempt_timeout_gen), | ||
| retry=None, | ||
| with self._metrics.create_operation( | ||
| OperationType.SAMPLE_ROW_KEYS | ||
| ) as operation_metric: | ||
|
|
||
| @CrossSync.convert | ||
| async def execute_rpc(): | ||
| results = await self.client._gapic_client.sample_row_keys( | ||
| request=SampleRowKeysRequest( | ||
| app_profile_id=self.app_profile_id, **self._request_path | ||
| ), | ||
| timeout=next(attempt_timeout_gen), | ||
| retry=None, | ||
| ) | ||
| return [(s.row_key, s.offset_bytes) async for s in results] | ||
|
|
||
| return await tracked_retry( | ||
| retry_fn=CrossSync.retry_target, | ||
| operation=operation_metric, | ||
| target=execute_rpc, | ||
| predicate=predicate, | ||
| timeout=operation_timeout, | ||
| ) | ||
|
daniel-sanche marked this conversation as resolved.
|
||
| return [(s.row_key, s.offset_bytes) async for s in results] | ||
|
|
||
| return await CrossSync.retry_target( | ||
| execute_rpc, | ||
| predicate, | ||
| sleep_generator, | ||
| operation_timeout, | ||
| exception_factory=_retry_exception_factory, | ||
| ) | ||
|
|
||
| @CrossSync.convert(replace_symbols={"MutationsBatcherAsync": "MutationsBatcher"}) | ||
| def mutations_batcher( | ||
|
|
@@ -1561,28 +1567,29 @@ async def mutate_row( | |
| # mutations should not be retried | ||
| predicate = retries.if_exception_type() | ||
|
|
||
| sleep_generator = retries.exponential_sleep_generator(0.01, 2, 60) | ||
|
|
||
| target = partial( | ||
| self.client._gapic_client.mutate_row, | ||
| request=MutateRowRequest( | ||
| row_key=row_key.encode("utf-8") | ||
| if isinstance(row_key, str) | ||
| else row_key, | ||
| mutations=[mutation._to_pb() for mutation in mutations_list], | ||
| app_profile_id=self.app_profile_id, | ||
| **self._request_path, | ||
| ), | ||
| timeout=attempt_timeout, | ||
| retry=None, | ||
| ) | ||
| return await CrossSync.retry_target( | ||
| target, | ||
| predicate, | ||
| sleep_generator, | ||
| operation_timeout, | ||
| exception_factory=_retry_exception_factory, | ||
| ) | ||
| with self._metrics.create_operation( | ||
| OperationType.MUTATE_ROW | ||
| ) as operation_metric: | ||
| target = partial( | ||
| self.client._gapic_client.mutate_row, | ||
| request=MutateRowRequest( | ||
| row_key=row_key.encode("utf-8") | ||
| if isinstance(row_key, str) | ||
| else row_key, | ||
| mutations=[mutation._to_pb() for mutation in mutations_list], | ||
| app_profile_id=self.app_profile_id, | ||
| **self._request_path, | ||
| ), | ||
| timeout=attempt_timeout, | ||
| retry=None, | ||
| ) | ||
| return await tracked_retry( | ||
| retry_fn=CrossSync.retry_target, | ||
| operation=operation_metric, | ||
| target=target, | ||
| predicate=predicate, | ||
| timeout=operation_timeout, | ||
| ) | ||
|
daniel-sanche marked this conversation as resolved.
|
||
|
|
||
| @CrossSync.convert | ||
| async def bulk_mutate_rows( | ||
|
|
@@ -1693,21 +1700,25 @@ async def check_and_mutate_row( | |
| ): | ||
| false_case_mutations = [false_case_mutations] | ||
| false_case_list = [m._to_pb() for m in false_case_mutations or []] | ||
| result = await self.client._gapic_client.check_and_mutate_row( | ||
| request=CheckAndMutateRowRequest( | ||
| true_mutations=true_case_list, | ||
| false_mutations=false_case_list, | ||
| predicate_filter=predicate._to_pb() if predicate is not None else None, | ||
| row_key=row_key.encode("utf-8") | ||
| if isinstance(row_key, str) | ||
| else row_key, | ||
| app_profile_id=self.app_profile_id, | ||
| **self._request_path, | ||
| ), | ||
| timeout=operation_timeout, | ||
| retry=None, | ||
| ) | ||
| return result.predicate_matched | ||
|
|
||
| with self._metrics.create_operation(OperationType.CHECK_AND_MUTATE): | ||
| result = await self.client._gapic_client.check_and_mutate_row( | ||
| request=CheckAndMutateRowRequest( | ||
| true_mutations=true_case_list, | ||
| false_mutations=false_case_list, | ||
| predicate_filter=predicate._to_pb() | ||
| if predicate is not None | ||
| else None, | ||
| row_key=row_key.encode("utf-8") | ||
| if isinstance(row_key, str) | ||
| else row_key, | ||
| app_profile_id=self.app_profile_id, | ||
| **self._request_path, | ||
| ), | ||
| timeout=operation_timeout, | ||
| retry=None, | ||
| ) | ||
| return result.predicate_matched | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. this is not wrapped in tracked_retry. Will the attempt level metrics (attempt latencies, server latencies and connectivity error count) still be recorded?
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. check_and_mutate shouldn't have retries, right? But yes, a sngle attempt is will be recorded when this completes. The duration/gfe_latency data is captured, and will be exported as those metrics in the follow-up PR |
||
|
|
||
| @CrossSync.convert | ||
| async def read_modify_write_row( | ||
|
|
@@ -1747,20 +1758,22 @@ async def read_modify_write_row( | |
| rules = [rules] | ||
| if not rules: | ||
| raise ValueError("rules must contain at least one item") | ||
| result = await self.client._gapic_client.read_modify_write_row( | ||
| request=ReadModifyWriteRowRequest( | ||
| rules=[rule._to_pb() for rule in rules], | ||
| row_key=row_key.encode("utf-8") | ||
| if isinstance(row_key, str) | ||
| else row_key, | ||
| app_profile_id=self.app_profile_id, | ||
| **self._request_path, | ||
| ), | ||
| timeout=operation_timeout, | ||
| retry=None, | ||
| ) | ||
| # construct Row from result | ||
| return Row._from_pb(result.row) | ||
|
|
||
| with self._metrics.create_operation(OperationType.READ_MODIFY_WRITE): | ||
| result = await self.client._gapic_client.read_modify_write_row( | ||
| request=ReadModifyWriteRowRequest( | ||
| rules=[rule._to_pb() for rule in rules], | ||
| row_key=row_key.encode("utf-8") | ||
| if isinstance(row_key, str) | ||
| else row_key, | ||
| app_profile_id=self.app_profile_id, | ||
| **self._request_path, | ||
| ), | ||
| timeout=operation_timeout, | ||
| retry=None, | ||
| ) | ||
| # construct Row from result | ||
| return Row._from_pb(result.row) | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. same question as check and mutate.
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Yes, those are also captured here |
||
|
|
||
| @CrossSync.convert | ||
| async def close(self): | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
You can’t perform that action at this time.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
where did sleep_generator and exception_factory go?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
tracked_retry contains them both. needed a custom versions of sleep_generator to report backoff, and exception_factory to report terminal errors metrics module. See go/bigtable-csm-python