use passed in project for all exporter tasks

Author: daniel-sanche

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

@@ -281,7 +281,6 @@ def __init__(
             try:
                 # create a metrics exporter using the same client configuration
                 exporter = BigtableMetricsExporter(
-                    project_id=self.project,
                     credentials=credentials,
                     client_options=client_options,
                 )

packages/google-cloud-bigtable/google/cloud/bigtable/data/_metrics/handlers/gcp_exporter.py

@@ -14,6 +14,7 @@
 
 from __future__ import annotations
 
+from collections import defaultdict
 import logging
 import time
 
@@ -136,18 +137,13 @@ class BigtableMetricsExporter(MetricExporter):
     We must use a custom exporter because the public one doesn't support writing to internal
     metrics like `bigtable.googleapis.com/internal/client/`
 
-    Each GoogleCloudMetricsHandler will maintain its own exporter instance associated with the
-    project_id it is configured with.
-
-    Args:
-        project_id: GCP project id to associate metrics with
+    Each GoogleCloudMetricsHandler will maintain its own exporter instance.
     """
 
-    def __init__(self, project_id: str, *client_args, **client_kwargs):
+    def __init__(self, *client_args, **client_kwargs):
         super().__init__()
         self.client = MetricServiceClient(*client_args, **client_kwargs)
         self.prefix = "bigtable.googleapis.com/internal/client"
-        self.project_id = project_id
 
     def export(
         self, metrics_data: MetricsData, timeout_millis: float = 10_000, **kwargs
@@ -158,19 +154,26 @@ def export(
         """
         deadline = time.monotonic() + (timeout_millis / 1000)
         metric_kind = MetricDescriptor.MetricKind.CUMULATIVE
-        all_series: list[TimeSeries] = []
+        series_by_project: dict[str, list[TimeSeries]] = defaultdict(list)
         # process each metric from OTel format into Cloud Monitoring format
         for resource_metric in metrics_data.resource_metrics:
             for scope_metric in resource_metric.scope_metrics:
                 for metric in scope_metric.metrics:
                     for data_point in metric.data.data_points:
                         if data_point.attributes:
+                            project_id = data_point.attributes.get(
+                                "resource_project", ""
+                            )
+                            if not project_id:
+                                _LOGGER.warning(
+                                    "Missing resource_project attribute for metric %s",
+                                    metric.name,
+                                )
+                                continue
                             monitored_resource = MonitoredResource(
                                 type="bigtable_client_raw",
                                 labels={
-                                    "project_id": data_point.attributes.get(
-                                        "resource_project", ""
-                                    ),
+                                    "project_id": project_id,
                                     "instance": data_point.attributes.get(
                                         "resource_instance", ""
                                     ),
@@ -209,15 +212,16 @@ def export(
                                 ),
                                 unit=metric.unit,
                             )
-                            all_series.append(series)
+                            series_by_project[project_id].append(series)
         # send all metrics to Cloud Monitoring
         try:
-            _LOGGER.debug(
-                "Exporting %d time series to Cloud Monitoring for project %s",
-                len(all_series),
-                self.project_id,
-            )
-            self._batch_write(all_series, deadline)
+            for project_id, series_list in series_by_project.items():
+                _LOGGER.debug(
+                    "Exporting %d time series to Cloud Monitoring for project %s",
+                    len(series_list),
+                    project_id,
+                )
+                self._batch_write(project_id, series_list, deadline)
             return MetricExportResult.SUCCESS
         except Exception as e:
             _LOGGER.warning(
@@ -226,13 +230,18 @@ def export(
             return MetricExportResult.FAILURE
 
     def _batch_write(
-        self, series: list[TimeSeries], deadline=None, max_batch_size=200
+        self,
+        project_id: str,
+        series: list[TimeSeries],
+        deadline=None,
+        max_batch_size=200,
     ) -> None:
         """
         Adapted from CloudMonitoringMetricsExporter
         https://github.com/GoogleCloudPlatform/opentelemetry-operations-python/blob/3668dfe7ce3b80dd01f42af72428de957b58b316/opentelemetry-exporter-gcp-monitoring/src/opentelemetry/exporter/cloud_monitoring/__init__.py#L82
 
         Args:
+            project_id: GCP project ID to write metrics to
             series: list of TimeSeries to write. Will be split into batches if necessary
             deadline: designates the time.time() at which to stop writing. If None, uses API default
             max_batch_size: maximum number of time series to write at once.
@@ -250,15 +259,15 @@ def _batch_write(
             batch = series[write_ind : write_ind + max_batch_size]
             self.client.create_service_time_series(
                 CreateTimeSeriesRequest(
-                    name=f"projects/{self.project_id}",
+                    name=f"projects/{project_id}",
                     time_series=batch,
                 ),
                 timeout=timeout,
             )
             _LOGGER.debug(
                 "Successfully wrote batch of %d time series to projects/%s",
                 len(batch),
-                self.project_id,
+                project_id,
             )
             write_ind += max_batch_size
 

packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/client.py

@@ -211,7 +211,6 @@ def __init__(
         else:
             try:
                 exporter = BigtableMetricsExporter(
-                    project_id=self.project,
                     credentials=credentials,
                     client_options=client_options,
                 )

packages/google-cloud-bigtable/tests/unit/data/_async/test_client.py

@@ -245,16 +245,10 @@ async def test_metrics_exporter_init_shares_arguments(
             use_emulator=False,
         ):
             mock_called.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(use_emulator=False) as client:
-            assert client._metrics.handlers[0]._exporter.project_id == client.project
-
     @CrossSync.pytest
     @mock.patch(
         "google.cloud.bigtable.data._async.client.BigtableMetricsExporter",

packages/google-cloud-bigtable/tests/unit/data/_metrics/test_gcp_exporter_handler.py

@@ -48,7 +48,7 @@ def _make_one(self, *args, **kwargs):
     def test_ctor_defaults(self, mock_auth):
         from google.cloud.bigtable import __version__ as CLIENT_VERSION
 
-        expected_exporter = BigtableMetricsExporter("project")
+        expected_exporter = BigtableMetricsExporter()
         with mock.patch.object(
             GoogleCloudMetricsHandler, "_generate_client_uid"
         ) as uid_mock:
@@ -64,7 +64,7 @@ def test_ctor_defaults(self, mock_auth):
     def test_ctor_explicit(self, mock_auth):
         expected_version = "my_version"
         expected_uid = "my_uid"
-        expected_exporter = BigtableMetricsExporter("project")
+        expected_exporter = BigtableMetricsExporter()
         handler = self._make_one(
             expected_exporter,
             client_uid=expected_uid,
@@ -133,22 +133,18 @@ def _make_one(self, *args, **kwargs):
     def test_ctor_defaults(self):
         from google.cloud.monitoring_v3 import MetricServiceClient
 
-        expected_project = "custom"
-        instance = self._make_one(expected_project)
-        assert instance.project_id == expected_project
+        instance = self._make_one()
         assert instance.prefix == "bigtable.googleapis.com/internal/client"
         assert isinstance(instance.client, MetricServiceClient)
 
     def test_ctor_mocks(self):
-        expected_project = "custom"
         with mock.patch(
             "google.cloud.monitoring_v3.MetricServiceClient.__init__",
             return_value=None,
         ) as mock_client:
             args = [mock.Mock(), object()]
             kwargs = {"a": "b"}
-            instance = self._make_one(expected_project, *args, **kwargs)
-            assert instance.project_id == expected_project
+            instance = self._make_one(*args, **kwargs)
             assert instance.prefix == "bigtable.googleapis.com/internal/client"
             mock_client.assert_called_once_with(*args, **kwargs)
 
@@ -161,7 +157,7 @@ def test_ctor_mocks(self):
     )
     def test__to_point_w_number(self, value, expected_field):
         """Test that NumberDataPoint is converted to a Point correctly."""
-        instance = self._make_one("project")
+        instance = self._make_one()
         expected_start_time_nanos = 100
         expected_end_time_nanos = 200
         dp = NumberDataPoint(
@@ -182,7 +178,7 @@ def test__to_point_w_number(self, value, expected_field):
 
     def test__to_point_w_histogram(self):
         """Test that HistogramDataPoint is converted to a Point correctly."""
-        instance = self._make_one("project")
+        instance = self._make_one()
         expected_start_time_nanos = 100
         expected_end_time_nanos = 200
         expected_count = 10
@@ -220,7 +216,7 @@ def test__to_point_w_histogram(self):
 
     def test__to_point_w_histogram_zero_count(self):
         """Test that HistogramDataPoint with zero count is converted to a Point correctly."""
-        instance = self._make_one("project")
+        instance = self._make_one()
         dp = HistogramDataPoint(
             attributes={},
             start_time_unix_nano=100,
@@ -252,15 +248,16 @@ def test__batch_write(
         self, num_series, batch_size, expected_calls, expected_batch_sizes
     ):
         """Test that _batch_write splits series into batches correctly."""
-        instance = self._make_one("project")
+        instance = self._make_one()
         instance.client = mock.Mock()
         series = [TimeSeries() for _ in range(num_series)]
-        instance._batch_write(series, max_batch_size=batch_size)
+        instance._batch_write("project", series, max_batch_size=batch_size)
         assert instance.client.create_service_time_series.call_count == expected_calls
         for i, call in enumerate(
             instance.client.create_service_time_series.call_args_list
         ):
             call_args, _ = call
+            assert call_args[0].name == "projects/project"
             assert len(call_args[0].time_series) == expected_batch_sizes[i]
 
     def test__batch_write_with_deadline(self):
@@ -269,12 +266,12 @@ def test__batch_write_with_deadline(self):
 
         from google.api_core import gapic_v1
 
-        instance = self._make_one("project")
+        instance = self._make_one()
         instance.client = mock.Mock()
         series = [TimeSeries() for _ in range(10)]
         # test with deadline
         deadline = time.monotonic() + 10
-        instance._batch_write(series, deadline=deadline)
+        instance._batch_write("project", series, deadline=deadline)
         (
             call_args,
             call_kwargs,
@@ -283,7 +280,7 @@ def test__batch_write_with_deadline(self):
         assert 9 < call_kwargs["timeout"] < 10
         # test without deadline
         instance.client.create_service_time_series.reset_mock()
-        instance._batch_write(series, deadline=None)
+        instance._batch_write("project", series, deadline=None)
         (
             call_args,
             call_kwargs,
@@ -294,7 +291,7 @@ def test__batch_write_with_deadline(self):
     def test_export(self):
         """Test that export correctly converts metrics and calls _batch_write."""
         project_id = "project"
-        instance = self._make_one(project_id)
+        instance = self._make_one()
         instance._batch_write = mock.Mock()
         # create mock metrics data
         expected_value = 123
@@ -334,7 +331,8 @@ def test_export(self):
         instance._batch_write.assert_called_once()
         # check the TimeSeries passed to _batch_write
         call_args, call_kwargs = instance._batch_write.call_args_list[0]
-        series_list = call_args[0]
+        assert call_args[0] == project_id
+        series_list = call_args[1]
         assert len(series_list) == 1
         series = series_list[0]
         assert series.metric.type == f"{instance.prefix}/operation_latencies"
@@ -352,7 +350,7 @@ def test_export(self):
 
     def test_export_no_attributes(self):
         """Test that export skips data points with no attributes."""
-        instance = self._make_one("project")
+        instance = self._make_one()
         instance._batch_write = mock.Mock()
         data_point = NumberDataPoint(
             attributes={}, start_time_unix_nano=100, time_unix_nano=200, value=123
@@ -376,18 +374,17 @@ def test_export_no_attributes(self):
         metrics_data = MetricsData(resource_metrics=[resource_metric])
         result = instance.export(metrics_data)
         assert result == MetricExportResult.SUCCESS
-        instance._batch_write.assert_called_once()
-        series_list = instance._batch_write.call_args[0][0]
-        assert len(series_list) == 0
+        instance._batch_write.assert_not_called()
 
     def test_exception_in_export(self):
         """
         make sure exceptions don't raise
         """
-        instance = self._make_one("project")
+        instance = self._make_one()
         instance._batch_write = mock.Mock(side_effect=Exception("test"))
         # create mock metrics data with one valid data point
         attributes = {
+            "resource_project": "project",
             "resource_instance": "instance1",
             "resource_cluster": "cluster1",
             "resource_table": "table1",

packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_client.py

@@ -216,15 +216,10 @@ def test_metrics_exporter_init_shares_arguments(
             use_emulator=False,
         ):
             mock_called.assert_called_once_with(
-                project_id=expected_project,
                 credentials=expected_credentials,
                 client_options=expected_options,
             )
 
-    def test_metrics_exporter_init_implicit_project(self):
-        with self._make_client(use_emulator=False) as client:
-            assert client._metrics.handlers[0]._exporter.project_id == client.project
-
     @mock.patch(
         "google.cloud.bigtable.data._async.client.BigtableMetricsExporter",
         side_effect=Exception("Auth error"),