From 93d855159c54a834a3ffc9510155db0469765b73 Mon Sep 17 00:00:00 2001 From: abhishekacharya200 Date: Mon, 12 Oct 2026 10:58:13 +0545 Subject: [PATCH 1/3] fix(bigtable): skip metrics export if previous write was too recent Cloud Monitoring rejects points written more often than the metric's 60s sample period. The final export on shutdown often came within 30s of a periodic export, so it failed and logged a warning with a traceback on normal process exit. Fixes #18576 --- .../data/_metrics/handlers/gcp_exporter.py | 20 ++++++ .../_metrics/test_gcp_exporter_handler.py | 62 +++++++++++++++++++ 2 files changed, 82 insertions(+) diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_metrics/handlers/gcp_exporter.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_metrics/handlers/gcp_exporter.py index dd96ce8eb01b..9d461d6c86b0 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_metrics/handlers/gcp_exporter.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_metrics/handlers/gcp_exporter.py @@ -164,6 +164,12 @@ def _get_monitoring_credentials( return credentials +# Cloud Monitoring rejects points written more often than the metric's sample +# period (60s for Bigtable client metrics). Exports that come sooner than half of +# that after the previous write (e.g. the final export on shutdown) are skipped +_MIN_EXPORT_INTERVAL_SECONDS = 30 + + class GoogleCloudMetricsHandler(OpenTelemetryMetricsHandler): """ Maintains an internal set of OpenTelemetry metrics for the Bigtable client library, @@ -228,6 +234,8 @@ def __init__(self, *client_args, **client_kwargs): # Once disabled, exports are skipped instead of failing (and logging) # on every export interval self._disabled = False + # time.monotonic() of the last successful write to Cloud Monitoring + self._last_export_time: float | None = None def export( self, metrics_data: MetricsData, timeout_millis: float = 10_000, **kwargs @@ -238,6 +246,17 @@ def export( """ if self._disabled: return MetricExportResult.FAILURE + if ( + self._last_export_time is not None + and time.monotonic() - self._last_export_time < _MIN_EXPORT_INTERVAL_SECONDS + ): + # writing again this soon would be rejected by Cloud Monitoring. + # Metrics are cumulative, so the next export still includes this data + _LOGGER.debug( + "Skipping metrics export: previous export was less than %d seconds ago", + _MIN_EXPORT_INTERVAL_SECONDS, + ) + return MetricExportResult.SUCCESS deadline = time.monotonic() + (timeout_millis / 1000) metric_kind = MetricDescriptor.MetricKind.CUMULATIVE series_by_project: dict[str, list[TimeSeries]] = defaultdict(list) @@ -291,6 +310,7 @@ def export( project_id, ) self._batch_write(project_id, series_list, deadline) + self._last_export_time = time.monotonic() return MetricExportResult.SUCCESS except (Unauthenticated, PermissionDenied) as e: # authentication/authorization failures won't resolve on their own. diff --git a/packages/google-cloud-bigtable/tests/unit/data/_metrics/test_gcp_exporter_handler.py b/packages/google-cloud-bigtable/tests/unit/data/_metrics/test_gcp_exporter_handler.py index 90e4691b1521..572739859601 100644 --- a/packages/google-cloud-bigtable/tests/unit/data/_metrics/test_gcp_exporter_handler.py +++ b/packages/google-cloud-bigtable/tests/unit/data/_metrics/test_gcp_exporter_handler.py @@ -129,6 +129,28 @@ def test_close(self): GoogleCloudMetricsHandler.close(mock_instance) assert mock_instance.meter_provider.shutdown.call_count == 1 + def test_close_soon_after_export(self): + """ + Closing the handler right after a periodic export should not write + to Cloud Monitoring again, since the write would be rejected for being + more frequent than the metric's sample period + + https://github.com/googleapis/google-cloud-python/issues/18576 + """ + with mock.patch("google.auth.default", return_value=(mock.Mock(), "project")): + exporter = BigtableMetricsExporter() + exporter.client = mock.Mock() + handler = self._make_one(exporter) + counter = handler.meter_provider.get_meter("test").create_counter("retry_count") + attributes = {"resource_project": "project", "resource_instance": "instance"} + counter.add(1, attributes) + # periodic export + assert handler.meter_provider.force_flush() + assert exporter.client.create_service_time_series.call_count == 1 + counter.add(1, attributes) + handler.close() + assert exporter.client.create_service_time_series.call_count == 1 + BIGTABLE_DATA_SCOPE = "https://www.googleapis.com/auth/bigtable.data" BIGTABLE_ADMIN_SCOPE = "https://www.googleapis.com/auth/bigtable.admin" @@ -676,3 +698,43 @@ def test_export_when_disabled(self): assert result == MetricExportResult.FAILURE to_point_mock.assert_not_called() instance._batch_write.assert_not_called() + + def test_export_skipped_if_previous_export_too_recent(self): + """ + Cloud Monitoring rejects points written more often than the sample period. + Exports within 30 seconds of the last successful write should be skipped + """ + instance = self._make_one() + instance.client = mock.Mock() + write_mock = instance.client.create_service_time_series + with mock.patch("time.monotonic", return_value=1000.0): + assert instance.export(self._make_metrics_data()) == ( + MetricExportResult.SUCCESS + ) + assert write_mock.call_count == 1 + # too soon: skipped without writing or failing + with mock.patch("time.monotonic", return_value=1029.0): + assert instance.export(self._make_metrics_data()) == ( + MetricExportResult.SUCCESS + ) + assert write_mock.call_count == 1 + # enough time has passed: written again + with mock.patch("time.monotonic", return_value=1030.0): + assert instance.export(self._make_metrics_data()) == ( + MetricExportResult.SUCCESS + ) + assert write_mock.call_count == 2 + + def test_export_not_skipped_after_failed_export(self): + """a failed write doesn't count as a recent export""" + instance = self._make_one() + instance.client = mock.Mock() + write_mock = instance.client.create_service_time_series + write_mock.side_effect = [api_exceptions.ServiceUnavailable("test"), None] + assert instance.export(self._make_metrics_data()) == ( + MetricExportResult.FAILURE + ) + assert instance.export(self._make_metrics_data()) == ( + MetricExportResult.SUCCESS + ) + assert write_mock.call_count == 2 From 58ef62aed2cb397ba3c5901426f56eb6c51b8a6a Mon Sep 17 00:00:00 2001 From: Abhishek Acharya Date: Sat, 17 Oct 2026 23:55:57 +1100 Subject: [PATCH 2/3] fix(bigtable): only record the export time after every project is written A failed write for a later project left the time set by an earlier one, so an immediate retry was skipped. Also close the handler in a finally block in test_close_soon_after_export. --- .../data/_metrics/handlers/gcp_exporter.py | 1 + .../_metrics/test_gcp_exporter_handler.py | 44 +++++++++++++++---- 2 files changed, 37 insertions(+), 8 deletions(-) diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_metrics/handlers/gcp_exporter.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_metrics/handlers/gcp_exporter.py index 9d461d6c86b0..27cf214b8cd5 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_metrics/handlers/gcp_exporter.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_metrics/handlers/gcp_exporter.py @@ -310,6 +310,7 @@ def export( project_id, ) self._batch_write(project_id, series_list, deadline) + if series_by_project: self._last_export_time = time.monotonic() return MetricExportResult.SUCCESS except (Unauthenticated, PermissionDenied) as e: diff --git a/packages/google-cloud-bigtable/tests/unit/data/_metrics/test_gcp_exporter_handler.py b/packages/google-cloud-bigtable/tests/unit/data/_metrics/test_gcp_exporter_handler.py index 572739859601..4508eb8de9c6 100644 --- a/packages/google-cloud-bigtable/tests/unit/data/_metrics/test_gcp_exporter_handler.py +++ b/packages/google-cloud-bigtable/tests/unit/data/_metrics/test_gcp_exporter_handler.py @@ -141,14 +141,21 @@ def test_close_soon_after_export(self): exporter = BigtableMetricsExporter() exporter.client = mock.Mock() handler = self._make_one(exporter) - counter = handler.meter_provider.get_meter("test").create_counter("retry_count") - attributes = {"resource_project": "project", "resource_instance": "instance"} - counter.add(1, attributes) - # periodic export - assert handler.meter_provider.force_flush() - assert exporter.client.create_service_time_series.call_count == 1 - counter.add(1, attributes) - handler.close() + try: + counter = handler.meter_provider.get_meter("test").create_counter( + "retry_count" + ) + attributes = { + "resource_project": "project", + "resource_instance": "instance", + } + counter.add(1, attributes) + # periodic export + assert handler.meter_provider.force_flush() + assert exporter.client.create_service_time_series.call_count == 1 + counter.add(1, attributes) + finally: + handler.close() assert exporter.client.create_service_time_series.call_count == 1 @@ -738,3 +745,24 @@ def test_export_not_skipped_after_failed_export(self): MetricExportResult.SUCCESS ) assert write_mock.call_count == 2 + + def test_export_not_skipped_after_partial_multi_project_failure(self): + """A partial failure across multiple projects doesn't count as a recent export""" + instance = self._make_one() + instance.client = mock.Mock() + write_mock = instance.client.create_service_time_series + write_mock.side_effect = [ + None, + api_exceptions.ServiceUnavailable("test"), + None, + None, + ] + + metrics_data = self._make_metrics_data(project_id="project1") + metrics_data_2 = self._make_metrics_data(project_id="project2") + metrics_data.resource_metrics.extend(metrics_data_2.resource_metrics) + + assert instance.export(metrics_data) == MetricExportResult.FAILURE + # Immediate retry (< 30s) is not skipped because the previous export failed + assert instance.export(metrics_data) == MetricExportResult.SUCCESS + assert write_mock.call_count == 4 From 3209ffacb6f97a84d6851d0bfdfad3ad08f9776a Mon Sep 17 00:00:00 2001 From: abhishekacharya200 Date: Thu, 22 Oct 2026 20:57:10 +1100 Subject: [PATCH 3/3] fix(bigtable): track the last metrics export time per project A recent write to one project no longer holds back exports to the others. Each project records its own last successful write, so a project that just exported skips the final flush while the rest still go through, and after a partial failure only the failed project is written again. --- .../data/_metrics/handlers/gcp_exporter.py | 34 ++++++++------- .../_metrics/test_gcp_exporter_handler.py | 43 ++++++++++++++++--- 2 files changed, 55 insertions(+), 22 deletions(-) diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_metrics/handlers/gcp_exporter.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_metrics/handlers/gcp_exporter.py index 27cf214b8cd5..9cf50e4a8a6e 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_metrics/handlers/gcp_exporter.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_metrics/handlers/gcp_exporter.py @@ -234,8 +234,9 @@ def __init__(self, *client_args, **client_kwargs): # Once disabled, exports are skipped instead of failing (and logging) # on every export interval self._disabled = False - # time.monotonic() of the last successful write to Cloud Monitoring - self._last_export_time: float | None = None + # time.monotonic() of the last successful write to Cloud Monitoring, per + # project, so a recent write to one project doesn't hold back the others + self._last_export_times: dict[str, float] = {} def export( self, metrics_data: MetricsData, timeout_millis: float = 10_000, **kwargs @@ -246,17 +247,6 @@ def export( """ if self._disabled: return MetricExportResult.FAILURE - if ( - self._last_export_time is not None - and time.monotonic() - self._last_export_time < _MIN_EXPORT_INTERVAL_SECONDS - ): - # writing again this soon would be rejected by Cloud Monitoring. - # Metrics are cumulative, so the next export still includes this data - _LOGGER.debug( - "Skipping metrics export: previous export was less than %d seconds ago", - _MIN_EXPORT_INTERVAL_SECONDS, - ) - return MetricExportResult.SUCCESS deadline = time.monotonic() + (timeout_millis / 1000) metric_kind = MetricDescriptor.MetricKind.CUMULATIVE series_by_project: dict[str, list[TimeSeries]] = defaultdict(list) @@ -304,14 +294,28 @@ def export( # send all metrics to Cloud Monitoring try: for project_id, series_list in series_by_project.items(): + last_export_time = self._last_export_times.get(project_id) + if ( + last_export_time is not None + and time.monotonic() - last_export_time + < _MIN_EXPORT_INTERVAL_SECONDS + ): + # writing again this soon would be rejected by Cloud Monitoring. + # Metrics are cumulative, so the next export still includes this data + _LOGGER.debug( + "Skipping metrics export for project %s: previous export " + "was less than %d seconds ago", + project_id, + _MIN_EXPORT_INTERVAL_SECONDS, + ) + continue _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) - if series_by_project: - self._last_export_time = time.monotonic() + self._last_export_times[project_id] = time.monotonic() return MetricExportResult.SUCCESS except (Unauthenticated, PermissionDenied) as e: # authentication/authorization failures won't resolve on their own. diff --git a/packages/google-cloud-bigtable/tests/unit/data/_metrics/test_gcp_exporter_handler.py b/packages/google-cloud-bigtable/tests/unit/data/_metrics/test_gcp_exporter_handler.py index 4508eb8de9c6..9d09aeda97e8 100644 --- a/packages/google-cloud-bigtable/tests/unit/data/_metrics/test_gcp_exporter_handler.py +++ b/packages/google-cloud-bigtable/tests/unit/data/_metrics/test_gcp_exporter_handler.py @@ -746,8 +746,11 @@ def test_export_not_skipped_after_failed_export(self): ) assert write_mock.call_count == 2 - def test_export_not_skipped_after_partial_multi_project_failure(self): - """A partial failure across multiple projects doesn't count as a recent export""" + def test_export_retries_only_failed_project_after_partial_failure(self): + """ + After a partial failure across projects, only the project that failed is + written again right away; the one that succeeded is too recent to rewrite + """ instance = self._make_one() instance.client = mock.Mock() write_mock = instance.client.create_service_time_series @@ -755,14 +758,40 @@ def test_export_not_skipped_after_partial_multi_project_failure(self): None, api_exceptions.ServiceUnavailable("test"), None, - None, ] metrics_data = self._make_metrics_data(project_id="project1") metrics_data_2 = self._make_metrics_data(project_id="project2") metrics_data.resource_metrics.extend(metrics_data_2.resource_metrics) - assert instance.export(metrics_data) == MetricExportResult.FAILURE - # Immediate retry (< 30s) is not skipped because the previous export failed - assert instance.export(metrics_data) == MetricExportResult.SUCCESS - assert write_mock.call_count == 4 + with mock.patch("time.monotonic", return_value=1000.0): + assert instance.export(metrics_data) == MetricExportResult.FAILURE + # Immediate retry (< 30s): project1 was just written, project2 failed + with mock.patch("time.monotonic", return_value=1010.0): + assert instance.export(metrics_data) == MetricExportResult.SUCCESS + assert write_mock.call_count == 3 + written = [c.args[0].name for c in write_mock.call_args_list] + assert written == [ + "projects/project1", + "projects/project2", + "projects/project2", + ] + + def test_recent_export_to_one_project_does_not_skip_another(self): + """the 30 second check is per project""" + instance = self._make_one() + instance.client = mock.Mock() + write_mock = instance.client.create_service_time_series + + with mock.patch("time.monotonic", return_value=1000.0): + assert instance.export(self._make_metrics_data(project_id="project1")) == ( + MetricExportResult.SUCCESS + ) + metrics_data = self._make_metrics_data(project_id="project1") + metrics_data_2 = self._make_metrics_data(project_id="project2") + metrics_data.resource_metrics.extend(metrics_data_2.resource_metrics) + # 10 seconds later: project1 is skipped, project2 is still written + with mock.patch("time.monotonic", return_value=1010.0): + assert instance.export(metrics_data) == MetricExportResult.SUCCESS + written = [c.args[0].name for c in write_mock.call_args_list] + assert written == ["projects/project1", "projects/project2"]