fix(spanner): export built-in metrics per project

packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/metrics_exporter.py

@@ -149,7 +149,6 @@ def _batch_write(self, series: List["TimeSeries"], timeout_millis: float) -> Non
         :param series: ProtoBuf TimeSeries
         :return:
         """
-        write_ind = 0
         timeout = timeout_millis / MILLIS_PER_SECOND
 
         retry = Retry(
@@ -166,21 +165,37 @@ def _batch_write(self, series: List["TimeSeries"], timeout_millis: float) -> Non
             deadline=timeout,
         )
 
-        while write_ind < len(series):
-            request = CreateTimeSeriesRequest(
-                name=self.project_name,
-                time_series=series[write_ind : write_ind + MAX_BATCH_WRITE],
-            )
-
-            try:
-                retry(self.client.create_service_time_series)(
-                    request=request,
-                    timeout=timeout,
+        # Group by resource project so mixed-project batches are not
+        # rejected (resource.labels.project_id must match name).
+        series_by_project: Dict[str, List["TimeSeries"]] = {}
+        for ts in series:
+            project_id = self.project_id
+            resource = getattr(ts, "resource", None)
+            labels = getattr(resource, "labels", None) if resource is not None else None
+            if labels:
+                label_project = labels.get("project_id")
+                if label_project:
+                    project_id = label_project
+            series_by_project.setdefault(project_id, []).append(ts)
+
+        for project_id, project_series in series_by_project.items():
+            project_name = self.client.common_project_path(project_id)
+            write_ind = 0
+            while write_ind < len(project_series):
+                request = CreateTimeSeriesRequest(
+                    name=project_name,
+                    time_series=project_series[write_ind : write_ind + MAX_BATCH_WRITE],
                 )
-            except Exception as e:
-                logger.error("Failed to export metrics to Cloud Monitoring: %s", e)
 
-            write_ind += MAX_BATCH_WRITE
+                try:
+                    retry(self.client.create_service_time_series)(
+                        request=request,
+                        timeout=timeout,
+                    )
+                except Exception as e:
+                    logger.error("Failed to export metrics to Cloud Monitoring: %s", e)
+
+                write_ind += MAX_BATCH_WRITE
 
     @staticmethod
     def _resource_to_monitored_resource_pb(

packages/google-cloud-spanner/tests/unit/test_metrics_exporter.py

@@ -382,6 +382,53 @@ def test_export(self):
             timeseries = batch_args[0]
             self.assertEqual(len(timeseries), 2)
 
+        def test_export_groups_time_series_by_project_id(self):
+            """Each project's series is sent in its own CreateTimeSeries request."""
+            from google.cloud.monitoring_v3 import MetricServiceClient
+
+            meter = self.provider.get_meter("gax-python")
+            counter = meter.create_counter(
+                name="attempt_count", description="A test counter", unit="count"
+            )
+
+            attributes_a = dict(self.metric_attributes)
+            attributes_a["project_id"] = "project-a"
+            attributes_b = dict(self.metric_attributes)
+            attributes_b["project_id"] = "project-b"
+            counter.add(1, attributes=attributes_a)
+            counter.add(1, attributes=attributes_b)
+
+            metrics = self.metric_reader.get_metrics_data()
+            mock_client = MagicMock(spec=MetricServiceClient)
+            mock_client.common_project_path.side_effect = lambda project: (
+                f"projects/{project}"
+            )
+            mock_client.create_service_time_series = Mock(return_value=None)
+
+            exporter = CloudMonitoringMetricsExporter("project-a", mock_client)
+            exporter.export(metrics)
+
+            calls = mock_client.create_service_time_series.call_args_list
+            self.assertEqual(len(calls), 2)
+
+            requests_by_name = {}
+            for call in calls:
+                request = call.kwargs.get("request")
+                if request is None and call.args:
+                    request = call.args[0]
+                self.assertIsNotNone(request)
+                requests_by_name[request.name] = request
+
+            self.assertEqual(
+                set(requests_by_name),
+                {"projects/project-a", "projects/project-b"},
+            )
+            for name, request in requests_by_name.items():
+                project_id = name.split("/")[-1]
+                self.assertGreater(len(request.time_series), 0)
+                for series in request.time_series:
+                    self.assertEqual(series.resource.labels["project_id"], project_id)
+
         def test_force_flush(self):
             """Verify that the unimplemented force flush can be called."""
             exporter = CloudMonitoringMetricsExporter(