added some unit tests

Author: daniel-sanche

google/cloud/bigtable/data/_async/client.py

@@ -984,7 +984,8 @@ def __init__(
                     exporter=client._gcp_metrics_exporter,
                     instance_id=instance_id,
                     table_id=table_id,
-                    app_profile_id=app_profile_id
+                    app_profile_id=app_profile_id,
+                    client_version=client._client_version(),
                 )
             ]
         )

google/cloud/bigtable/data/_metrics/handlers/opentelemetry.py

@@ -26,6 +26,8 @@
 from google.cloud.bigtable.data._metrics.data_model import CompletedAttemptMetric
 from google.cloud.bigtable.data._metrics.data_model import CompletedOperationMetric
 
+# conversion factor for converting from nanoseconds to milliseconds
+NS_TO_MS= 1e6
 
 class _OpenTelemetryInstruments:
     """
@@ -124,14 +126,15 @@ def __init__(
         table_id: str,
         app_profile_id: str | None = None,
         client_uid: str | None = None,
+        client_version: str | None = None,
         instruments: _OpenTelemetryInstruments = _OpenTelemetryInstruments(),
-        **kwargs,
     ):
         super().__init__()
         self.otel = instruments
+        client_version = client_version or bigtable_version
         # fixed labels sent with each metric update
         self.shared_labels = {
-            "client_name": f"python-bigtable/{bigtable_version}",
+            "client_name": f"python-bigtable/{client_version}",
             "client_uid": client_uid or self._generate_client_uid(),
             "resource_instance": instance_id,
             "resource_table": table_id,
@@ -162,6 +165,7 @@ def on_operation_complete(self, op: CompletedOperationMetric) -> None:
         Update the metrics associated with a completed operation:
           - operation_latencies
           - retry_count
+          - first_response_latencies
         """
         labels = {
             "method": op.op_type.value,
@@ -173,14 +177,14 @@ def on_operation_complete(self, op: CompletedOperationMetric) -> None:
         is_streaming = str(op.is_streaming)
 
         self.otel.operation_latencies.record(
-            op.duration_ns / 1e6, {"streaming": is_streaming, **labels}
+            op.duration_ns / NS_TO_MS, {"streaming": is_streaming, **labels}
         )
         if (
             op.op_type == OperationType.READ_ROWS
             and op.first_response_latency_ns is not None
         ):
             self.otel.first_response_latencies.record(
-                op.first_response_latency_ns / 1e6, labels
+                op.first_response_latency_ns / NS_TO_MS, labels
             )
         # only record completed attempts if there were retries
         if op.completed_attempts:
@@ -192,7 +196,6 @@ def on_attempt_complete(
         """
         Update the metrics associated with a completed attempt:
           - attempt_latencies
-          - first_response_latencies
           - server_latencies
           - connectivity_error_count
           - application_latencies
@@ -208,19 +211,19 @@ def on_attempt_complete(
         is_streaming = str(op.is_streaming)
 
         self.otel.attempt_latencies.record(
-            attempt.duration_ns / 1e6, {"streaming": is_streaming, "status": status, **labels}
+            attempt.duration_ns / NS_TO_MS, {"streaming": is_streaming, "status": status, **labels}
         )
-        combined_throttling = attempt.grpc_throttling_time_ns / 1e6
+        combined_throttling = attempt.grpc_throttling_time_ns / NS_TO_MS
         if not op.completed_attempts:
             # add flow control latency to first attempt's throttling latency
-            combined_throttling += (op.flow_throttling_time_ns / 1e6 if op.flow_throttling_time_ns else 0)
+            combined_throttling += (op.flow_throttling_time_ns / NS_TO_MS if op.flow_throttling_time_ns else 0)
         self.otel.throttling_latencies.record(combined_throttling, labels)
         self.otel.application_latencies.record(
-            (attempt.application_blocking_time_ns + attempt.backoff_before_attempt_ns) / 1e6, labels
+            (attempt.application_blocking_time_ns + attempt.backoff_before_attempt_ns) / NS_TO_MS, labels
         )
         if attempt.gfe_latency_ns is not None:
             self.otel.server_latencies.record(
-                attempt.gfe_latency_ns / 1e6,
+                attempt.gfe_latency_ns / NS_TO_MS,
                 {"streaming": is_streaming, "status": status, **labels},
             )
         else:

tests/system/data/test_system_async.py

@@ -1381,11 +1381,10 @@ def test_metric_existence(self, client, table_id, metrics_client, time_interval,
             """
             Checks existence of each metric in Cloud Monitoring
             """
-            from google.cloud.bigtable import __version__ as CLIENT_VERSION
             for m in methods:
                 metric_filter = (
                     f'metric.type = "bigtable.googleapis.com/client/{metric}" ' +
-                    f'AND metric.labels.client_name = "python-bigtable/{CLIENT_VERSION}" '
+                    f'AND metric.labels.client_name = "python-bigtable/{client._client_version()}" ' +
                     f'AND resource.labels.table = "{table_id}" '
                 )
                 results = list(metrics_client.list_time_series(

tests/unit/data/_async/test_client.py

@@ -110,6 +110,10 @@ def _make_client(cls, *args, use_emulator=True, **kwargs):
 
     @CrossSync.pytest
     async def test_ctor(self):
+        from google.cloud.bigtable.data._metrics.handlers.gcp_exporter import (
+            BigtableMetricsExporter,
+        )
+
         expected_project = "project-id"
         expected_credentials = AnonymousCredentials()
         client = self._make_client(
@@ -123,6 +127,8 @@ async def test_ctor(self):
         assert client._channel_refresh_task is not None
         assert client.transport._credentials == expected_credentials
         assert isinstance(client._metrics_interceptor, CrossSync.MetricsInterceptor)
+        assert client._gcp_metrics_exporter is not None
+        assert isinstance(client._gcp_metrics_exporter, BigtableMetricsExporter)
         await client.close()
 
     @CrossSync.pytest
@@ -189,6 +195,22 @@ async def test_ctor_dict_options(self):
             start_background_refresh.assert_called_once()
             await client.close()
 
+    @CrossSync.pytest
+    async def test_metrics_exporter_init_shares_arguments(self):
+        expected_credentials = AnonymousCredentials()
+        expected_project = "custom_project"
+        expected_options = client_options.ClientOptions()
+        expected_options.credentials_file = None
+        expected_options.quota_project_id = None
+        with mock.patch("google.cloud.bigtable.data._metrics.handlers.gcp_exporter.BigtableMetricsExporter.__init__", return_value=None) as exporter_mock:
+            async with self._make_client(project=expected_project, credentials=expected_credentials, client_options=expected_options):
+                exporter_mock.assert_called_once_with(project_id=expected_project, credentials=expected_credentials, client_options=expected_options)
+
+    @CrossSync.pytest
+    async def test_metrics_exporter_init_implicit_project(self):
+        async with self._make_client() as client:
+            assert client._gcp_metrics_exporter.project_id == client.project
+
     @CrossSync.pytest
     async def test_veneer_grpc_headers(self):
         client_component = "data-async" if CrossSync.is_async else "data"
@@ -1163,6 +1185,9 @@ async def test_ctor(self):
         from google.cloud.bigtable.data._metrics import (
             BigtableClientSideMetricsController,
         )
+        from google.cloud.bigtable.data._metrics import (
+            GoogleCloudMetricsHandler
+        )
 
         expected_table_id = "table-id"
         expected_instance_id = "instance-id"
@@ -1205,6 +1230,8 @@ async def test_ctor(self):
         assert instance_key in client._active_instances
         assert client._instance_owners[instance_key] == {id(table)}
         assert isinstance(table._metrics, BigtableClientSideMetricsController)
+        assert len(table._metrics.handlers) == 1
+        assert isinstance(table._metrics.handlers[0], GoogleCloudMetricsHandler)
         assert table.default_operation_timeout == expected_operation_timeout
         assert table.default_attempt_timeout == expected_attempt_timeout
         assert (
@@ -1495,7 +1522,7 @@ def _make_one(
     async def test_ctor(self):
         from google.cloud.bigtable.data._helpers import _WarmedInstanceKey
         from google.cloud.bigtable.data._metrics import (
-            BigtableClientSideMetricsController,
+            BigtableClientSideMetricsController, GoogleCloudMetricsHandler
         )
 
         expected_table_id = "table-id"
@@ -1546,6 +1573,8 @@ async def test_ctor(self):
         assert instance_key in client._active_instances
         assert client._instance_owners[instance_key] == {id(view)}
         assert isinstance(view._metrics, BigtableClientSideMetricsController)
+        assert len(view._metrics.handlers) == 1
+        assert isinstance(view._metrics.handlers[0], GoogleCloudMetricsHandler)
         assert view.default_operation_timeout == expected_operation_timeout
         assert view.default_attempt_timeout == expected_attempt_timeout
         assert (