feat(async-publisher): degrade AsyncPubSubPublisher to noop when GCP credentials are absent

CHANGELOG.md

@@ -31,7 +31,7 @@ For more information about each release including git tags and artifacts, see [R
 - Event stream shows sensible defaults so first-load isn't empty ([#297](https://github.com/roostorg/osprey/pull/297) by [@haileyok](https://github.com/haileyok))
 - Replace `react-scripts` with `rsbuild`/`rspack` for UI builds ([#235](https://github.com/roostorg/osprey/pull/235) by [@chimosky](https://github.com/chimosky))
 - Migrate from npm to pnpm via Corepack ([#252](https://github.com/roostorg/osprey/pull/252) by [@haileyok](https://github.com/haileyok))
-- `PubSubPublisher` degrades to a noop when GCP credentials are absent or when `DISABLE_GCP_PUBSUB` is set, so Osprey runs without GCP config instead of failing ([#388](https://github.com/roostorg/osprey/pull/388) by [@julietshen](https://github.com/julietshen))
+- Pub/Sub publishers degrade to a noop when GCP credentials are absent or when `DISABLE_GCP_PUBSUB` is set, so Osprey runs without GCP config instead of failing: `PubSubPublisher` ([#388](https://github.com/roostorg/osprey/pull/388)) and `AsyncPubSubPublisher` ([#411](https://github.com/roostorg/osprey/pull/411)) (by [@julietshen](https://github.com/julietshen))
 
 ### Fixed
 

osprey_async_worker/src/osprey/async_worker/lib/publisher.py

@@ -9,6 +9,7 @@
 
 from google.api_core.retry import Retry
 from google.cloud import pubsub_v1
+from osprey.worker.lib.gcp_credentials import gcp_credentials_available, gcp_pubsub_disabled
 from osprey.worker.lib.instruments import metrics
 from pydantic import BaseModel
 
@@ -29,6 +30,12 @@ class AsyncPubSubPublisher:
 
     Messages are buffered in an asyncio.Queue and flushed either when
     the batch reaches max_messages or after max_latency_seconds.
+
+    Degrades to noop mode when the DISABLE_GCP_PUBSUB env var is set, or when GCP
+    credentials cannot be resolved at construction time (e.g. local dev or adopter
+    environments without GCP). In noop mode no underlying client is built and the
+    publish paths return immediately. A one-time warning is logged at construction
+    so the inert state is visible.
     """
 
     def __init__(
@@ -39,15 +46,31 @@ def __init__(
         max_latency_seconds: float = 1.0,
     ):
         self._topic_path = f'projects/{project_id}/topics/{topic_id}'
+        self._metric_tags = [f'project:{project_id}', f'topic:{topic_id}']
+        self._flush_task: asyncio.Task[None] | None = None
+        if gcp_pubsub_disabled():
+            self._enabled = False
+            logger.warning(
+                'DISABLE_GCP_PUBSUB is set, AsyncPubSubPublisher disabled (project=%s, topic=%s)',
+                project_id,
+                topic_id,
+            )
+            return
+        self._enabled = gcp_credentials_available()
+        if not self._enabled:
+            logger.warning(
+                'GCP credentials not detected, AsyncPubSubPublisher running in noop mode (project=%s, topic=%s)',
+                project_id,
+                topic_id,
+            )
+            return
         self._client = pubsub_v1.PublisherClient(
             batch_settings=pubsub_v1.types.BatchSettings(max_messages=1),
         )
         self._queue: asyncio.Queue[bytes] = asyncio.Queue()
         self._max_messages = max_messages
         self._max_latency = max_latency_seconds
-        self._flush_task: asyncio.Task[None] | None = None
         self._started = False
-        self._metric_tags = [f'project:{project_id}', f'topic:{topic_id}']
 
     def _ensure_started(self) -> None:
         """Start the background flush task on first publish."""
@@ -129,6 +152,8 @@ def publish(self, data: BaseModel) -> None:
 
     def publish_bytes(self, data: bytes) -> None:
         """Queue raw bytes for async batched publishing."""
+        if not self._enabled:
+            return
         self._ensure_started()
         try:
             self._queue.put_nowait(data)
@@ -138,6 +163,8 @@ def publish_bytes(self, data: bytes) -> None:
 
     async def stop(self) -> None:
         """Flush remaining messages and stop."""
+        if not self._enabled:
+            return
         if self._flush_task is not None:
             self._flush_task.cancel()
             try:

osprey_async_worker/src/osprey/async_worker/tests/test_publisher.py

@@ -2,13 +2,18 @@
 
 from unittest.mock import MagicMock, patch
 
+import pytest
 from google.api_core.exceptions import NotFound
+from osprey.async_worker.lib import publisher as publisher_module
 from osprey.async_worker.lib.publisher import _PUBLISH_RETRY, AsyncPubSubPublisher
 
 
 def _make_publisher():
-    """Return an AsyncPubSubPublisher with a mocked PublisherClient."""
-    with patch('osprey.async_worker.lib.publisher.pubsub_v1.PublisherClient'):
+    """Return an enabled AsyncPubSubPublisher with a mocked PublisherClient."""
+    with (
+        patch('osprey.async_worker.lib.publisher.gcp_credentials_available', return_value=True),
+        patch('osprey.async_worker.lib.publisher.pubsub_v1.PublisherClient'),
+    ):
         publisher = AsyncPubSubPublisher(project_id='proj', topic_id='topic')
     publisher._client = MagicMock()
     return publisher
@@ -48,3 +53,50 @@ def test_permanent_failure_metric_fires(mock_metrics):
     assert len(failure_calls) == 1
     assert failure_calls[0][0][0] == 'async_pubsub_publisher.publish.failure'
     assert f'error:{exc.__class__.__name__}' in failure_calls[0][1]['tags']
+
+
+def test_noops_when_creds_absent(caplog: pytest.LogCaptureFixture) -> None:
+    with (
+        patch('osprey.async_worker.lib.publisher.gcp_credentials_available', return_value=False),
+        patch('osprey.async_worker.lib.publisher.pubsub_v1.PublisherClient') as client_cls,
+    ):
+        with caplog.at_level('WARNING', logger=publisher_module.logger.name):
+            publisher = AsyncPubSubPublisher(project_id='proj', topic_id='topic')
+    assert publisher._enabled is False
+    assert not client_cls.called
+    assert not hasattr(publisher, '_client')
+    assert 'noop mode' in caplog.text
+    assert 'project=proj' in caplog.text
+    assert 'topic=topic' in caplog.text
+
+
+def test_disabled_via_env(monkeypatch, caplog: pytest.LogCaptureFixture) -> None:
+    monkeypatch.setenv('DISABLE_GCP_PUBSUB', 'true')
+    with (
+        patch('osprey.async_worker.lib.publisher.gcp_credentials_available') as cred_check,
+        patch('osprey.async_worker.lib.publisher.pubsub_v1.PublisherClient') as client_cls,
+    ):
+        with caplog.at_level('WARNING', logger=publisher_module.logger.name):
+            publisher = AsyncPubSubPublisher(project_id='proj', topic_id='topic')
+    assert publisher._enabled is False
+    assert not client_cls.called
+    # The opt-out short-circuits before probing credentials.
+    assert not cred_check.called
+    assert 'DISABLE_GCP_PUBSUB' in caplog.text
+
+
+@patch('osprey.async_worker.lib.publisher.metrics')
+def test_publish_bytes_short_circuits_silently_when_disabled(mock_metrics) -> None:
+    with patch('osprey.async_worker.lib.publisher.gcp_credentials_available', return_value=False):
+        publisher = AsyncPubSubPublisher(project_id='proj', topic_id='topic')
+        publisher.publish_bytes(b'data')
+
+    mock_metrics.increment.assert_not_called()
+
+
+async def test_stop_short_circuits_when_disabled() -> None:
+    with patch('osprey.async_worker.lib.publisher.gcp_credentials_available', return_value=False):
+        publisher = AsyncPubSubPublisher(project_id='proj', topic_id='topic')
+        await publisher.stop()
+
+    assert not hasattr(publisher, '_client')

osprey_worker/src/osprey/worker/lib/data_exporters/test/conftest.py

@@ -28,5 +28,5 @@ def pubsub_client_mock(monkeypatch_session, monkeypatch) -> MagicMock:
     # monkeypatch so it reverts after each test instead of leaking into later
     # modules (e.g. test_publisher.py, whose disabled-publisher tests would then
     # build a real client and hang on stop()).
-    monkeypatch.setattr(publisher, '_check_gcp_credentials', lambda: True)
+    monkeypatch.setattr(publisher, 'gcp_credentials_available', lambda: True)
     return pubsub_client_mock

osprey_worker/src/osprey/worker/lib/gcp_credentials.py

@@ -0,0 +1,36 @@
+"""Shared detection of whether GCP Pub/Sub publishing should be active.
+
+Used by the Pub/Sub publishers to decide whether to run for real or degrade to
+a noop (e.g. local dev or adopter environments without GCP). Publishing is off
+when DISABLE_GCP_PUBSUB is set, or when GCP credentials cannot be resolved. The
+credential result is cached at module level so google.auth.default() is probed
+at most once per process, and guarded by a lock so concurrent constructors
+probe only once.
+"""
+
+import os
+import threading
+
+import google.auth
+from google.auth.exceptions import DefaultCredentialsError
+
+_gcp_credentials_available: bool | None = None
+_gcp_credentials_lock = threading.Lock()
+
+
+def gcp_pubsub_disabled() -> bool:
+    return os.environ.get('DISABLE_GCP_PUBSUB', '').lower() == 'true'
+
+
+def gcp_credentials_available() -> bool:
+    global _gcp_credentials_available
+    if _gcp_credentials_available is None:
+        with _gcp_credentials_lock:
+            # Re-check under the lock so concurrent callers probe google.auth.default() only once.
+            if _gcp_credentials_available is None:
+                try:
+                    google.auth.default()
+                    _gcp_credentials_available = True
+                except DefaultCredentialsError:
+                    _gcp_credentials_available = False
+    return _gcp_credentials_available

osprey_worker/src/osprey/worker/lib/publisher.py

@@ -1,12 +1,9 @@
 import abc
 import logging
-import os
-import threading
 from typing import TypeVar
 
-import google.auth
-from google.auth.exceptions import DefaultCredentialsError
 from google.cloud import pubsub_v1
+from osprey.worker.lib.gcp_credentials import gcp_credentials_available, gcp_pubsub_disabled
 from osprey.worker.lib.instruments import metrics
 from osprey.worker.lib.pubsub.publisher_client import BatchPubsubPublisherClient
 from pydantic import BaseModel
@@ -72,15 +69,15 @@ def __init__(
         self._project = project_id
         self._raise_on_error = raise_on_error
         self._tags = [f'project:{self._project}', f'topic:{self._topic_name}']
-        if _gcp_pubsub_disabled():
+        if gcp_pubsub_disabled():
             self._enabled = False
             logger.warning(
                 'DISABLE_GCP_PUBSUB is set, PubSubPublisher disabled (project=%s, topic=%s)',
                 project_id,
                 topic_id,
             )
             return
-        self._enabled = _check_gcp_credentials()
+        self._enabled = gcp_credentials_available()
         if not self._enabled:
             logger.warning(
                 'GCP credentials not detected, PubSubPublisher running in noop mode (project=%s, topic=%s)',
@@ -139,25 +136,3 @@ def publish(self, data: str, attributes: dict[str, str] | None = None) -> None:
             attributes = {}
 
         super().publish(data, attributes)  # type: ignore[type-var]
-
-
-def _gcp_pubsub_disabled() -> bool:
-    return os.environ.get('DISABLE_GCP_PUBSUB', '').lower() == 'true'
-
-
-_gcp_credentials_available: bool | None = None
-_gcp_credentials_lock = threading.Lock()
-
-
-def _check_gcp_credentials() -> bool:
-    global _gcp_credentials_available
-    if _gcp_credentials_available is None:
-        with _gcp_credentials_lock:
-            # Re-check under the lock so concurrent constructors probe google.auth.default() only once.
-            if _gcp_credentials_available is None:
-                try:
-                    google.auth.default()
-                    _gcp_credentials_available = True
-                except DefaultCredentialsError:
-                    _gcp_credentials_available = False
-    return _gcp_credentials_available

osprey_worker/src/osprey/worker/lib/tests/test_gcp_credentials.py

@@ -0,0 +1,41 @@
+from typing import Iterator
+from unittest.mock import MagicMock, patch
+
+import pytest
+from google.auth.exceptions import DefaultCredentialsError
+from osprey.worker.lib import gcp_credentials
+from osprey.worker.lib.gcp_credentials import gcp_credentials_available, gcp_pubsub_disabled
+
+
+@pytest.fixture(autouse=True)
+def reset_cred_cache() -> Iterator[None]:
+    gcp_credentials._gcp_credentials_available = None
+    yield
+    gcp_credentials._gcp_credentials_available = None
+
+
+def test_true_when_default_resolves() -> None:
+    with patch.object(gcp_credentials.google.auth, 'default', return_value=(MagicMock(), 'proj')):
+        assert gcp_credentials_available() is True
+
+
+def test_false_when_default_raises() -> None:
+    with patch.object(gcp_credentials.google.auth, 'default', side_effect=DefaultCredentialsError()):
+        assert gcp_credentials_available() is False
+
+
+def test_result_is_cached() -> None:
+    with patch.object(gcp_credentials.google.auth, 'default', return_value=(MagicMock(), 'proj')) as default_mock:
+        gcp_credentials_available()
+        gcp_credentials_available()
+        gcp_credentials_available()
+    assert default_mock.call_count == 1
+
+
+def test_pubsub_disabled_reads_env(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv('DISABLE_GCP_PUBSUB', raising=False)
+    assert gcp_pubsub_disabled() is False
+    monkeypatch.setenv('DISABLE_GCP_PUBSUB', 'true')
+    assert gcp_pubsub_disabled() is True
+    monkeypatch.setenv('DISABLE_GCP_PUBSUB', 'false')
+    assert gcp_pubsub_disabled() is False

osprey_worker/src/osprey/worker/lib/tests/test_publisher.py

@@ -1,40 +1,13 @@
-from typing import Iterator
 from unittest.mock import MagicMock, patch
 
 import pytest
-from google.auth.exceptions import DefaultCredentialsError
 from osprey.worker.lib import publisher
-from osprey.worker.lib.publisher import PubSubPublisher, _check_gcp_credentials
-
-
-@pytest.fixture(autouse=True)
-def reset_cred_cache() -> Iterator[None]:
-    publisher._gcp_credentials_available = None
-    yield
-    publisher._gcp_credentials_available = None
-
-
-def test_check_gcp_credentials_true_when_default_resolves() -> None:
-    with patch.object(publisher.google.auth, 'default', return_value=(MagicMock(), 'proj')):
-        assert _check_gcp_credentials() is True
-
-
-def test_check_gcp_credentials_false_when_default_raises() -> None:
-    with patch.object(publisher.google.auth, 'default', side_effect=DefaultCredentialsError()):
-        assert _check_gcp_credentials() is False
-
-
-def test_check_gcp_credentials_is_cached() -> None:
-    with patch.object(publisher.google.auth, 'default', return_value=(MagicMock(), 'proj')) as default_mock:
-        _check_gcp_credentials()
-        _check_gcp_credentials()
-        _check_gcp_credentials()
-    assert default_mock.call_count == 1
+from osprey.worker.lib.publisher import PubSubPublisher
 
 
 def test_pubsub_publisher_constructs_client_when_creds_present() -> None:
     with (
-        patch.object(publisher.google.auth, 'default', return_value=(MagicMock(), 'proj')),
+        patch.object(publisher, 'gcp_credentials_available', return_value=True),
         patch.object(publisher, 'BatchPubsubPublisherClient') as client_cls,
     ):
         pub = PubSubPublisher('proj', 'topic')
@@ -44,7 +17,7 @@ def test_pubsub_publisher_constructs_client_when_creds_present() -> None:
 
 def test_pubsub_publisher_noops_when_creds_absent(caplog: pytest.LogCaptureFixture) -> None:
     with (
-        patch.object(publisher.google.auth, 'default', side_effect=DefaultCredentialsError()),
+        patch.object(publisher, 'gcp_credentials_available', return_value=False),
         patch.object(publisher, 'BatchPubsubPublisherClient') as client_cls,
     ):
         with caplog.at_level('WARNING', logger=publisher.logger.name):
@@ -61,7 +34,7 @@ def test_pubsub_publisher_disabled_via_env(
 ) -> None:
     monkeypatch.setenv('DISABLE_GCP_PUBSUB', 'true')
     with (
-        patch.object(publisher, '_check_gcp_credentials') as cred_check,
+        patch.object(publisher, 'gcp_credentials_available') as cred_check,
         patch.object(publisher, 'BatchPubsubPublisherClient') as client_cls,
     ):
         with caplog.at_level('WARNING', logger=publisher.logger.name):
@@ -75,7 +48,7 @@ def test_pubsub_publisher_disabled_via_env(
 
 def test_publish_short_circuits_silently_when_disabled() -> None:
     with (
-        patch.object(publisher.google.auth, 'default', side_effect=DefaultCredentialsError()),
+        patch.object(publisher, 'gcp_credentials_available', return_value=False),
         patch.object(publisher, 'metrics') as metrics_mock,
     ):
         pub = PubSubPublisher('proj', 'topic')
@@ -84,7 +57,7 @@ def test_publish_short_circuits_silently_when_disabled() -> None:
 
 
 def test_stop_short_circuits_when_disabled() -> None:
-    with patch.object(publisher.google.auth, 'default', side_effect=DefaultCredentialsError()):
+    with patch.object(publisher, 'gcp_credentials_available', return_value=False):
         pub = PubSubPublisher('proj', 'topic')
         pub.stop()
     assert not hasattr(pub, '_publisher')